You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2026/01/08 13:22:37 UTC
[PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
loserwang1024 opened a new pull request, #2331:
URL: https://github.com/apache/fluss/pull/2331
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #2330
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #2331:
URL: https://github.com/apache/fluss/pull/2331#discussion_r2675507496
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -133,69 +138,118 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
}
}
- private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+ private List<TableChange> checkAndFilterDuplicateTableChanges(
+ TablePath tablePath,
+ List<TableChange> tableChanges,
+ org.apache.fluss.metadata.Schema flussSchema)
throws TableNotExistException {
+ if (tableChanges.isEmpty()) {
+ return tableChanges;
+ }
+
try {
+ // Get current Paimon table schema
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
- Schema currentSchema = fileStoreTable.schema().toSchema();
-
- for (TableChange change : tableChanges) {
- if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- if (!isColumnAlreadyExists(currentSchema, addColumn)) {
- return true;
- }
- } else {
- return true;
+ List<DataField> paimonFields =
+ fileStoreTable.schema().toSchema().fields().stream()
+ .filter(field -> !SYSTEM_COLUMNS.containsKey(field.name()))
+ .collect(Collectors.toList());
+ List<org.apache.fluss.metadata.Schema.Column> flussColumns = flussSchema.getColumns();
+
+ if (paimonFields.size() < flussColumns.size()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Paimon table has less columns (%d) than Fluss schema (%d)",
+ paimonFields.size(), flussColumns.size()));
+ }
+
+ // Validate schema compatibility
+ validateExistingColumns(paimonFields, flussColumns);
+
+ if (paimonFields.size() == flussColumns.size()) {
+ return tableChanges;
+ }
+
+ // 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.
+ 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.",
+ paimonFields.size(), flussColumns.size(), tableChanges);
+ if (flussColumns.size() + tableChanges.size() != paimonFields.size()) {
+ throw new InvalidAlterTableException(errorMsg);
+ }
+
+ 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);
Review Comment:
Maybe we could split the checks and provide distinct error messages to help users more easily identify issues related to the Paimon schema.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #2331:
URL: https://github.com/apache/fluss/pull/2331#discussion_r2693500453
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -133,69 +138,118 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
}
}
- private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+ private List<TableChange> checkAndFilterDuplicateTableChanges(
+ TablePath tablePath,
+ List<TableChange> tableChanges,
+ org.apache.fluss.metadata.Schema flussSchema)
throws TableNotExistException {
+ if (tableChanges.isEmpty()) {
+ return tableChanges;
+ }
+
try {
+ // Get current Paimon table schema
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
- Schema currentSchema = fileStoreTable.schema().toSchema();
-
- for (TableChange change : tableChanges) {
- if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- if (!isColumnAlreadyExists(currentSchema, addColumn)) {
- return true;
- }
- } else {
- return true;
+ List<DataField> paimonFields =
+ fileStoreTable.schema().toSchema().fields().stream()
+ .filter(field -> !SYSTEM_COLUMNS.containsKey(field.name()))
+ .collect(Collectors.toList());
+ List<org.apache.fluss.metadata.Schema.Column> flussColumns = flussSchema.getColumns();
+
+ if (paimonFields.size() < flussColumns.size()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Paimon table has less columns (%d) than Fluss schema (%d)",
+ paimonFields.size(), flussColumns.size()));
+ }
+
+ // Validate schema compatibility
+ validateExistingColumns(paimonFields, flussColumns);
+
+ if (paimonFields.size() == flussColumns.size()) {
+ return tableChanges;
+ }
+
+ // 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.
+ 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.",
+ paimonFields.size(), flussColumns.size(), tableChanges);
+ if (flussColumns.size() + tableChanges.size() != paimonFields.size()) {
+ throw new InvalidAlterTableException(errorMsg);
+ }
+
+ 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()
Review Comment:
@beryllw
> It’s not related to the existing Paimon schema
If paimon schema is same as fluss schema, all the operations(both add column and set config) can be accepted. Otherwise, only add columns can be accepted to make the schema same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #2331:
URL: https://github.com/apache/fluss/pull/2331#discussion_r2675489067
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -133,69 +138,118 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
}
}
- private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+ private List<TableChange> checkAndFilterDuplicateTableChanges(
+ TablePath tablePath,
+ List<TableChange> tableChanges,
+ org.apache.fluss.metadata.Schema flussSchema)
throws TableNotExistException {
+ if (tableChanges.isEmpty()) {
+ return tableChanges;
+ }
+
try {
+ // Get current Paimon table schema
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
- Schema currentSchema = fileStoreTable.schema().toSchema();
-
- for (TableChange change : tableChanges) {
- if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- if (!isColumnAlreadyExists(currentSchema, addColumn)) {
- return true;
- }
- } else {
- return true;
+ List<DataField> paimonFields =
+ fileStoreTable.schema().toSchema().fields().stream()
+ .filter(field -> !SYSTEM_COLUMNS.containsKey(field.name()))
+ .collect(Collectors.toList());
+ List<org.apache.fluss.metadata.Schema.Column> flussColumns = flussSchema.getColumns();
+
+ if (paimonFields.size() < flussColumns.size()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Paimon table has less columns (%d) than Fluss schema (%d)",
+ paimonFields.size(), flussColumns.size()));
+ }
+
+ // Validate schema compatibility
+ validateExistingColumns(paimonFields, flussColumns);
+
+ if (paimonFields.size() == flussColumns.size()) {
+ return tableChanges;
+ }
+
+ // 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.
+ 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.",
+ paimonFields.size(), flussColumns.size(), tableChanges);
+ if (flussColumns.size() + tableChanges.size() != paimonFields.size()) {
+ throw new InvalidAlterTableException(errorMsg);
+ }
+
+ 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()
Review Comment:
Maybe we could perform this check earlier—it’s not related to the existing Paimon schema. Also, it’s already validated here:
https://github.com/apache/fluss/blob/256bd7f3838a72965720eb2a8c6c7c0575f46b07/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java#L101-L104
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2331:
URL: https://github.com/apache/fluss/pull/2331#issuecomment-3815052115
@loserwang1024 the test coverage of `PaimonLakeCatalog` is failed, could you run `PaimonLakeCatalogTest` with coverage and improve the line coverage of `PaimonLakeCatalog` by adding tests `PaimonLakeCatalogTest`?
```
Warning: Rule violated for class org.apache.fluss.lake.paimon.PaimonLakeCatalog: lines covered ratio is 0.57, but expected minimum is 0.70
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #2331:
URL: https://github.com/apache/fluss/pull/2331#issuecomment-3815314946
@wuchong some old codes are remained, I have removed them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2331:
URL: https://github.com/apache/fluss/pull/2331#discussion_r2672377900
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -133,69 +138,118 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
}
}
- private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+ private List<TableChange> checkAndFilterDuplicateTableChanges(
+ TablePath tablePath,
+ List<TableChange> tableChanges,
+ org.apache.fluss.metadata.Schema flussSchema)
throws TableNotExistException {
+ if (tableChanges.isEmpty()) {
+ return tableChanges;
+ }
+
try {
+ // Get current Paimon table schema
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
- Schema currentSchema = fileStoreTable.schema().toSchema();
-
- for (TableChange change : tableChanges) {
- if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- if (!isColumnAlreadyExists(currentSchema, addColumn)) {
- return true;
- }
- } else {
- return true;
+ List<DataField> paimonFields =
+ fileStoreTable.schema().toSchema().fields().stream()
+ .filter(field -> !SYSTEM_COLUMNS.containsKey(field.name()))
+ .collect(Collectors.toList());
+ List<org.apache.fluss.metadata.Schema.Column> flussColumns = flussSchema.getColumns();
+
+ if (paimonFields.size() < flussColumns.size()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Paimon table has less columns (%d) than Fluss schema (%d)",
+ paimonFields.size(), flussColumns.size()));
+ }
+
+ // Validate schema compatibility
+ validateExistingColumns(paimonFields, flussColumns);
+
+ if (paimonFields.size() == flussColumns.size()) {
+ return tableChanges;
+ }
+
+ // 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.
+ 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.",
Review Comment:
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.
```suggestion
"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.",
```
##########
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java:
##########
@@ -84,5 +85,12 @@ interface Context {
/** Get the fluss principal currently accessing the catalog. */
FlussPrincipal getFlussPrincipal();
+
+ /**
+ * Get the current schema of fluss.
+ *
+ * @return the current schema of fluss,
Review Comment:
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.
```suggestion
* @return the current schema of fluss,
* @since 0.10
```
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -133,69 +138,118 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
}
}
- private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+ private List<TableChange> checkAndFilterDuplicateTableChanges(
+ TablePath tablePath,
+ List<TableChange> tableChanges,
+ org.apache.fluss.metadata.Schema flussSchema)
throws TableNotExistException {
+ if (tableChanges.isEmpty()) {
+ return tableChanges;
+ }
+
try {
+ // Get current Paimon table schema
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
- Schema currentSchema = fileStoreTable.schema().toSchema();
-
- for (TableChange change : tableChanges) {
- if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- if (!isColumnAlreadyExists(currentSchema, addColumn)) {
- return true;
- }
- } else {
- return true;
+ List<DataField> paimonFields =
+ fileStoreTable.schema().toSchema().fields().stream()
+ .filter(field -> !SYSTEM_COLUMNS.containsKey(field.name()))
+ .collect(Collectors.toList());
+ List<org.apache.fluss.metadata.Schema.Column> flussColumns = flussSchema.getColumns();
+
+ if (paimonFields.size() < flussColumns.size()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Paimon table has less columns (%d) than Fluss schema (%d)",
+ paimonFields.size(), flussColumns.size()));
+ }
+
+ // Validate schema compatibility
+ validateExistingColumns(paimonFields, flussColumns);
+
+ if (paimonFields.size() == flussColumns.size()) {
+ return tableChanges;
+ }
+
+ // 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.
+ 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.",
+ paimonFields.size(), flussColumns.size(), tableChanges);
+ if (flussColumns.size() + tableChanges.size() != paimonFields.size()) {
+ throw new InvalidAlterTableException(errorMsg);
+ }
+
+ 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);
}
}
Review Comment:
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.
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -133,69 +138,118 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
}
}
- private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+ private List<TableChange> checkAndFilterDuplicateTableChanges(
+ TablePath tablePath,
+ List<TableChange> tableChanges,
+ org.apache.fluss.metadata.Schema flussSchema)
throws TableNotExistException {
+ if (tableChanges.isEmpty()) {
+ return tableChanges;
+ }
+
try {
+ // Get current Paimon table schema
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
- Schema currentSchema = fileStoreTable.schema().toSchema();
-
- for (TableChange change : tableChanges) {
- if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- if (!isColumnAlreadyExists(currentSchema, addColumn)) {
- return true;
- }
- } else {
- return true;
+ List<DataField> paimonFields =
+ fileStoreTable.schema().toSchema().fields().stream()
+ .filter(field -> !SYSTEM_COLUMNS.containsKey(field.name()))
+ .collect(Collectors.toList());
+ List<org.apache.fluss.metadata.Schema.Column> flussColumns = flussSchema.getColumns();
+
+ if (paimonFields.size() < flussColumns.size()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Paimon table has less columns (%d) than Fluss schema (%d)",
+ paimonFields.size(), flussColumns.size()));
+ }
+
+ // Validate schema compatibility
Review Comment:
There's an extra space in the comment. Should be "Validate schema compatibility" instead of "Validate schema compatibility".
```suggestion
// Validate schema compatibility
```
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -133,69 +138,118 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
}
}
- private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+ private List<TableChange> checkAndFilterDuplicateTableChanges(
+ TablePath tablePath,
+ List<TableChange> tableChanges,
+ org.apache.fluss.metadata.Schema flussSchema)
throws TableNotExistException {
+ if (tableChanges.isEmpty()) {
+ return tableChanges;
+ }
+
try {
+ // Get current Paimon table schema
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
- Schema currentSchema = fileStoreTable.schema().toSchema();
-
- for (TableChange change : tableChanges) {
- if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- if (!isColumnAlreadyExists(currentSchema, addColumn)) {
- return true;
- }
- } else {
- return true;
+ List<DataField> paimonFields =
+ fileStoreTable.schema().toSchema().fields().stream()
+ .filter(field -> !SYSTEM_COLUMNS.containsKey(field.name()))
+ .collect(Collectors.toList());
+ List<org.apache.fluss.metadata.Schema.Column> flussColumns = flussSchema.getColumns();
+
+ if (paimonFields.size() < flussColumns.size()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Paimon table has less columns (%d) than Fluss schema (%d)",
+ paimonFields.size(), flussColumns.size()));
+ }
+
+ // Validate schema compatibility
+ validateExistingColumns(paimonFields, flussColumns);
+
+ if (paimonFields.size() == flussColumns.size()) {
+ return tableChanges;
+ }
+
+ // 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.
Review Comment:
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".
```suggestion
// 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.
```
##########
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java:
##########
@@ -84,5 +85,12 @@ interface Context {
/** Get the fluss principal currently accessing the catalog. */
FlussPrincipal getFlussPrincipal();
+
+ /**
+ * Get the current schema of fluss.
+ *
+ * @return the current schema of fluss,
Review Comment:
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".
```suggestion
* Get the current schema of Fluss.
*
* @return the current schema of Fluss.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2331:
URL: https://github.com/apache/fluss/pull/2331
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] Compare paimon schema and Fluss schema before alter table. [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2331:
URL: https://github.com/apache/fluss/pull/2331#issuecomment-3777083732
@loserwang1024 as we discussed offline, we can refactor the implementation to reuse the logic of `PaimonTableValidation#validatePaimonSchemaCompatible` instead of checking fields by fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org