You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "LiebingYu (via GitHub)" <gi...@apache.org> on 2025/10/21 06:42:49 UTC
[PR] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
LiebingYu opened a new pull request, #1847:
URL: https://github.com/apache/fluss/pull/1847
<!--
*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: #846
This PR only tolerate Paimon lake table existent if the schema and properties matches. Other lake storage should implement thier own logic to tolerate lake table existent.
<!-- 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] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1847:
URL: https://github.com/apache/fluss/pull/1847#discussion_r2517960050
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -142,6 +167,67 @@ private void alterTable(Identifier tablePath, List<SchemaChange> tableChanges)
}
}
+ private void validatePaimonSchemaCapability(
+ Identifier tablePath, Schema existingSchema, Schema newSchema) {
+ // Adjust options for comparison
+ Map<String, String> existingOptions = new HashMap<>(existingSchema.options());
+ Map<String, String> newOptions = new HashMap<>(newSchema.options());
+ // `path` will be set automatically by Paimon, so we need to remove it in existing options
+ existingOptions.remove(CoreOptions.PATH.key());
+ // when enable datalake with an existing table, `table.datalake.enabled` will be `false`
+ // in existing options, but `true` in new options.
+ String datalakeConfigKey = FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_ENABLED.key();
+ if (Boolean.FALSE.toString().equalsIgnoreCase(existingOptions.get(datalakeConfigKey))) {
+ existingOptions.remove(datalakeConfigKey);
+ newOptions.remove(datalakeConfigKey);
+ }
+
+ // Build schemas with adjusted options for comparison
+ Schema adjustedExistingSchema = buildSchemaWithOptions(existingSchema, existingOptions);
+ Schema adjustedNewSchema = buildSchemaWithOptions(newSchema, newOptions);
+
+ if (!adjustedExistingSchema.equals(adjustedNewSchema)) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The table %s already exists in Paimon catalog, but the table schema is not compatible. "
+ + "Existing schema: %s, new schema: %s. "
+ + "Please first drop the table in Paimon catalog or use a new table name.",
+ tablePath.getEscapedFullName(),
+ adjustedExistingSchema,
+ adjustedNewSchema));
+ }
+ }
+
+ private Schema buildSchemaWithOptions(Schema schema, Map<String, String> options) {
+ Schema.Builder builder = Schema.newBuilder();
+ // Copy fields
+ for (org.apache.paimon.types.DataField field : schema.fields()) {
+ builder.column(field.name(), field.type(), field.description());
Review Comment:
The `field.description()` is copied to the new schema builder, but Paimon's DataField may have nullable information that should also be preserved. Consider verifying if the field's nullable attribute is correctly maintained when rebuilding the schema, or document why description-only copying is sufficient for schema comparison.
```suggestion
builder.column(field.name(), field.type(), field.description(), field.nullable());
```
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java:
##########
@@ -410,38 +406,31 @@ private void preAlterTableProperties(
+ " in data lake, because the Fluss cluster doesn't enable datalake tables.");
}
- boolean isLakeTableNewlyCreated = false;
// to enable lake table
if (!isDataLakeEnabled(tableDescriptor)) {
// before create table in fluss, we may create in lake
try {
lakeCatalog.createTable(tablePath, newDescriptor, lakeCatalogContext);
- // no need to alter lake table if it is newly created
- isLakeTableNewlyCreated = true;
} catch (TableAlreadyExistException e) {
- // TODO: should tolerate if the lake exist but matches our schema. This ensures
- // eventually consistent by idempotently creating the table multiple times. See
- // #846
- throw new LakeTableAlreadyExistException(
- String.format(
- "The table %s already exists in %s catalog, please "
- + "first drop the table in %s catalog or use a new table name.",
- tablePath, dataLakeFormat, dataLakeFormat));
+ throw new LakeTableAlreadyExistException(e.getMessage(), e);
}
}
+ }
- // only need to alter lake table if it is not newly created
- if (!isLakeTableNewlyCreated) {
- {
- try {
- lakeCatalog.alterTable(tablePath, tableChanges, lakeCatalogContext);
- } catch (TableNotExistException e) {
- throw new FlussRuntimeException(
- "Lake table doesn't exists for lake-enabled table "
- + tablePath
- + ", which shouldn't be happened. Please check if the lake table was deleted manually.",
- e);
- }
+ // We should always alter lake table even though datalake is disabled.
+ // Otherwise, if user alter the fluss table when datalake is disabled, then enable datalake
+ // again, the lake table will mismatch.
+ if (lakeCatalog != null) {
+ try {
+ lakeCatalog.alterTable(tablePath, tableChanges, lakeCatalogContext);
+ } catch (TableNotExistException e) {
+ // only throw TableNotExistException if datalake is enabled
+ if (isDataLakeEnabled(newDescriptor)) {
+ throw new FlussRuntimeException(
+ "Lake table doesn't exists for lake-enabled table "
Review Comment:
Corrected spelling of 'doesn't exists' to 'doesn't exist'.
```suggestion
"Lake table doesn't exist for lake-enabled table "
```
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java:
##########
@@ -610,6 +860,84 @@ void testAlterLakeEnabledTableProperties() throws Exception {
admin.alterTable(TablePath.of(DATABASE, "not_exist_table"), tableChanges, true).get();
}
+ @Test
+ void testEnableLakeTableAfterAlterTableProperties() throws Exception {
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("k1", "v1");
+ customProperties.put("paimon.file.format", "parquet");
+
+ // create table
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .build())
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, false)
+ .customProperties(customProperties)
+ .distributedBy(BUCKET_NUM, "c1", "c2")
+ .build();
+ TablePath tablePath = TablePath.of(DATABASE, "enable_lake_table_after_alter_properties");
+ admin.createTable(tablePath, tableDescriptor, false).get();
+ // paimon table should not exist because lake table is disable
+ assertThatThrownBy(
+ () ->
+ paimonCatalog.getTable(
+ Identifier.create(DATABASE, tablePath.getTableName())))
+ .isInstanceOf(Catalog.TableNotExistException.class)
+ .hasMessage(String.format("Table %s does not exist.", tablePath));
+
+ // alter fluss table properties
+ List<TableChange> tableChanges =
+ Arrays.asList(TableChange.reset("k1"), TableChange.set("k2", "v2"));
+ admin.alterTable(tablePath, tableChanges, false).get();
+ // enable lake table should be ok
+ TableChange.SetOption enableLake =
+ TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
+ admin.alterTable(tablePath, Collections.singletonList(enableLake), false).get();
+ Table paimonTable =
+ paimonCatalog.getTable(Identifier.create(DATABASE, tablePath.getTableName()));
+ customProperties.remove("k1");
+ customProperties.put("k2", "v2");
+ Map<String, String> newProperties = new HashMap<>(tableDescriptor.getProperties());
+ newProperties.put(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
+ tableDescriptor = tableDescriptor.withProperties(newProperties, customProperties);
+ verifyPaimonTable(
+ paimonTable,
+ tableDescriptor,
+ RowType.of(
+ new DataType[] {
+ org.apache.paimon.types.DataTypes.INT(),
+ org.apache.paimon.types.DataTypes.STRING(),
+ // for __bucket, __offset, __timestamp
+ org.apache.paimon.types.DataTypes.INT(),
+ org.apache.paimon.types.DataTypes.BIGINT(),
+ org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()
+ },
+ new String[] {
+ "c1",
+ "c2",
+ BUCKET_COLUMN_NAME,
+ OFFSET_COLUMN_NAME,
+ TIMESTAMP_COLUMN_NAME
+ }),
+ "c1,c2",
+ BUCKET_NUM);
+
+ // disable lake table
+ TableChange.SetOption disableLake =
+ TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false");
+ admin.alterTable(tablePath, Collections.singletonList(disableLake), false).get();
+
+ // alter fluss table properties when lake table is disabled
+ tableChanges = Collections.singletonList(TableChange.set("k2", "v22"));
+ admin.alterTable(tablePath, tableChanges, false).get();
+
+ // enable lake table again should be ok, even though the table properties are changed
Review Comment:
Corrected grammar in comment: 'even though the table properties are changed' should be 'even though the table properties have changed'.
```suggestion
// enable lake table again should be ok, even though the table properties have changed
```
--
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] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1847:
URL: https://github.com/apache/fluss/pull/1847#discussion_r2514037538
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableManagerITCase.java:
##########
@@ -103,9 +103,7 @@ void testCreateAndGetTable() throws Exception {
lakeTablePath, lakeTableDescriptor, false))
.get())
.cause()
- .isInstanceOf(LakeTableAlreadyExistException.class)
- .hasMessage(
- "The table %s already exists in paimon catalog, please first drop the table in paimon catalog or use a new table name.",
- lakeTablePath);
+ .isInstanceOf(TableAlreadyExistException.class)
+ .hasMessage("Table %s already exists.", lakeTablePath);
Review Comment:
I think still we need a clear exception to tell user the table exists in which catalog and guide users how to handle the exception.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java:
##########
@@ -400,7 +399,6 @@ private void preAlterTableProperties(
TableDescriptor newDescriptor,
List<TableChange> tableChanges,
LakeCatalog lakeCatalog,
- DataLakeFormat dataLakeFormat,
Review Comment:
we can also remove `dataLakeFormat` in `alterTableProperties`
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java:
##########
@@ -294,13 +292,7 @@ public CompletableFuture<CreateTableResponse> createTable(CreateTableRequest req
tableDescriptor,
new DefaultLakeCatalogContext(currentSession().getPrincipal()));
} catch (TableAlreadyExistException e) {
- throw new LakeTableAlreadyExistException(
- String.format(
- "The table %s already exists in %s catalog, please "
- + "first drop the table in %s catalog or use a new table name.",
- tablePath,
- lakeCatalogContainer.getDataLakeFormat(),
- lakeCatalogContainer.getDataLakeFormat()));
+ throw new LakeTableAlreadyExistException(e.getMessage(), e);
Review Comment:
why change this?
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -120,7 +123,19 @@ private void createTable(Identifier tablePath, Schema schema)
// not ignore if table exists
paimonCatalog.createTable(tablePath, schema, false);
} catch (Catalog.TableAlreadyExistException e) {
- throw new TableAlreadyExistException("Table " + tablePath + " already exists.");
+ try {
+ Table table = paimonCatalog.getTable(tablePath);
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ validatePaimonSchemaCapability(fileStoreTable.schema().toSchema(), schema);
+ } catch (Catalog.TableNotExistException tableNotExistException) {
+ throw new RuntimeException(
Review Comment:
nit:
add comment
`// shouldn't happen in normal cases`
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonSchemaValidation.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.paimon.utils;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.exception.TableAlreadyExistException;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.schema.Schema;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.fluss.lake.paimon.utils.PaimonConversions.FLUSS_CONF_PREFIX;
+
+/** Validator of Paimon table schema. */
+public class PaimonSchemaValidation {
+
+ public static void validatePaimonSchemaCapability(Schema existingSchema, Schema newSchema) {
Review Comment:
Can just be use `existingSchema.equals(newSchema)` with some adjust to options?
Then, I think we can remove PaimonSchemaValidation class
--
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] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
Posted by "LiebingYu (via GitHub)" <gi...@apache.org>.
LiebingYu commented on code in PR #1847:
URL: https://github.com/apache/fluss/pull/1847#discussion_r2516574608
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java:
##########
@@ -294,13 +292,7 @@ public CompletableFuture<CreateTableResponse> createTable(CreateTableRequest req
tableDescriptor,
new DefaultLakeCatalogContext(currentSession().getPrincipal()));
} catch (TableAlreadyExistException e) {
- throw new LakeTableAlreadyExistException(
- String.format(
- "The table %s already exists in %s catalog, please "
- + "first drop the table in %s catalog or use a new table name.",
- tablePath,
- lakeCatalogContainer.getDataLakeFormat(),
- lakeCatalogContainer.getDataLakeFormat()));
+ throw new LakeTableAlreadyExistException(e.getMessage(), e);
Review Comment:
After we support compatibility with existing lake tables, I think the exception messages here should provide more details to help users quickly locate issues. For example, whether it is a schema mismatch or the table is not empty. I suggest that the specific exception information should be delegated to LakeCatalog.
--
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] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1847:
URL: https://github.com/apache/fluss/pull/1847#discussion_r2518010513
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java:
##########
@@ -474,16 +732,8 @@ void testAlterLakeEnabledLogTable() throws Exception {
// try to enable lake table again
Review Comment:
Enhance the test with the most common case:
1: write some data
2: disable datalake
3: enable datalake again, verify data still exist. we can just verify snapshot for clean code.
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -142,6 +167,67 @@ private void alterTable(Identifier tablePath, List<SchemaChange> tableChanges)
}
}
+ private void validatePaimonSchemaCapability(
+ Identifier tablePath, Schema existingSchema, Schema newSchema) {
+ // Adjust options for comparison
+ Map<String, String> existingOptions = new HashMap<>(existingSchema.options());
Review Comment:
I'm wondering whether the following code is better:
```
private void validatePaimonSchemaCapability(
Identifier tablePath, Schema existingSchema, Schema newSchema) {
// Adjust options for comparison
Map<String, String> existingOptions = existingSchema.options();
Map<String, String> newOptions =newSchema.options();
// `path` will be set automatically by Paimon, so we need to remove it in existing options
existingOptions.remove(CoreOptions.PATH.key());
// when enable datalake with an existing table, `table.datalake.enabled` will be `false`
// in existing options, but `true` in new options.
String datalakeConfigKey = FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_ENABLED.key();
if (Boolean.FALSE.toString().equalsIgnoreCase(existingOptions.get(datalakeConfigKey))) {
existingOptions.remove(datalakeConfigKey);
newOptions.remove(datalakeConfigKey);
}
if (!existingSchema.equals(newSchema)) {
throw new TableAlreadyExistException(
String.format(
"The table %s already exists in Paimon catalog, but the table schema is not compatible. "
+ "Existing schema: %s, new schema: %s. "
+ "Please first drop the table in Paimon catalog or use a new table name.",
tablePath.getEscapedFullName(),
existingSchema,
newSchema));
}
}
```
which modify schema's option directly.
I'm afraid that paimon schema introduce another field and we forget set the field in method `buildSchemaWithOptions`
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java:
##########
@@ -401,6 +408,257 @@ void testCreateLakeEnableTableWithUnsettablePaimonOptions() {
}
}
+ @Test
+ void testCreateLakeEnableTableWithExistLakeTable() throws Exception {
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("k1", "v1");
+ customProperties.put("paimon.file.format", "parquet");
+
+ // 1. test for existing lake table without bucket keys
+ TableDescriptor logTableWithoutBucketKeys =
+ createTableDescriptor(
+ 2,
+ BUCKET_NUM,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ customProperties,
+ false);
+ TablePath logTablePathWithoutBucketkeys =
+ TablePath.of(DATABASE, "log_table_without_bucket_keys");
+ admin.createTable(logTablePathWithoutBucketkeys, logTableWithoutBucketKeys, false).get();
+ // drop fluss table, lake table should still exist
+ admin.dropTable(logTablePathWithoutBucketkeys, false).get();
+ // create the same fluss table again should be ok
+ admin.createTable(logTablePathWithoutBucketkeys, logTableWithoutBucketKeys, false).get();
+ admin.dropTable(logTablePathWithoutBucketkeys, false).get();
+
+ // paimon table use dynamic bucket for fluss log table without bucket keys
+ // so it should be ok to create the same fluss table with a new bucket num
+ logTableWithoutBucketKeys = logTableWithoutBucketKeys.withBucketCount(BUCKET_NUM + 1);
+ admin.createTable(logTablePathWithoutBucketkeys, logTableWithoutBucketKeys, false).get();
+ admin.dropTable(logTablePathWithoutBucketkeys, false).get();
+
+ // create log table with bucket keys will throw exception
+ TableDescriptor logTableWithoutBucketKeys1 =
+ createTableDescriptor(
+ 2,
+ BUCKET_NUM,
+ Arrays.asList("c1", "c2"),
+ Collections.emptyList(),
+ customProperties,
+ false);
+ assertThatThrownBy(
+ () ->
+ admin.createTable(
+ logTablePathWithoutBucketkeys,
+ logTableWithoutBucketKeys1,
+ false)
+ .get())
+ .cause()
+ .isInstanceOf(LakeTableAlreadyExistException.class)
+ .hasMessage(
+ "The table `fluss`.`log_table_without_bucket_keys` already exists in Paimon catalog, but the table schema is not compatible. "
+ + "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=-1, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, file.format=parquet, fluss.k1=v1}, comment=null}, "
+ + "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=3, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, bucket-key=c1,c2, file.format=parquet, fluss.k1=v1}, comment=null}. "
+ + "Please first drop the table in Paimon catalog or use a new table name.");
+
+ // create log table with different fields will throw exception
+ TableDescriptor logTableWithoutBucketKeys2 =
+ createTableDescriptor(
+ 3,
+ BUCKET_NUM,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ customProperties,
+ false);
+ assertThatThrownBy(
+ () ->
+ admin.createTable(
+ logTablePathWithoutBucketkeys,
+ logTableWithoutBucketKeys2,
+ false)
+ .get())
+ .cause()
+ .isInstanceOf(LakeTableAlreadyExistException.class)
+ .hasMessage(
+ "The table `fluss`.`log_table_without_bucket_keys` already exists in Paimon catalog, but the table schema is not compatible. "
+ + "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=-1, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, file.format=parquet, fluss.k1=v1}, comment=null}, "
+ + "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `c3` STRING, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=-1, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, file.format=parquet, fluss.k1=v1}, comment=null}. "
+ + "Please first drop the table in Paimon catalog or use a new table name.");
+
+ // create table with primary keys will throw exception
Review Comment:
I think verify the basic cases should be fine. Not need to verify all case in here since the code path should be same.
```
void testCreateLakeEnableTableWithExistLakeTable() throws Exception {
Map<String, String> customProperties = new HashMap<>();
customProperties.put("k1", "v1");
customProperties.put("paimon.file.format", "parquet");
// 1. test for existing lake table without bucket keys
TableDescriptor logTableWithoutBucketKeys =
createTableDescriptor(
2,
BUCKET_NUM,
Collections.emptyList(),
Collections.emptyList(),
customProperties,
false);
TablePath logTablePathWithoutBucketkeys =
TablePath.of(DATABASE, "log_table_without_bucket_keys");
admin.createTable(logTablePathWithoutBucketkeys, logTableWithoutBucketKeys, false).get();
// drop fluss table, lake table should still exist
admin.dropTable(logTablePathWithoutBucketkeys, false).get();
// create the same fluss table again should be ok
admin.createTable(logTablePathWithoutBucketkeys, logTableWithoutBucketKeys, false).get();
admin.dropTable(logTablePathWithoutBucketkeys, false).get();
// paimon table use dynamic bucket for fluss log table without bucket keys
// so it should be ok to create the same fluss table with a new bucket num
logTableWithoutBucketKeys = logTableWithoutBucketKeys.withBucketCount(BUCKET_NUM + 1);
admin.createTable(logTablePathWithoutBucketkeys, logTableWithoutBucketKeys, false).get();
admin.dropTable(logTablePathWithoutBucketkeys, false).get();
// create log table with bucket keys will throw exception
TableDescriptor logTableWithoutBucketKeys1 =
createTableDescriptor(
2,
BUCKET_NUM,
Arrays.asList("c1", "c2"),
Collections.emptyList(),
customProperties,
false);
assertThatThrownBy(
() ->
admin.createTable(
logTablePathWithoutBucketkeys,
logTableWithoutBucketKeys1,
false)
.get())
.cause()
.isInstanceOf(LakeTableAlreadyExistException.class)
.hasMessage(
"The table `fluss`.`log_table_without_bucket_keys` already exists in Paimon catalog, but the table schema is not compatible. "
+ "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=-1, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, file.format=parquet, fluss.k1=v1}, comment=null}, "
+ "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=3, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, bucket-key=c1,c2, file.format=parquet, fluss.k1=v1}, comment=null}. "
+ "Please first drop the table in Paimon catalog or use a new table name.");
// create log table with different fields will throw exception
TableDescriptor logTableWithoutBucketKeys2 =
createTableDescriptor(
3,
BUCKET_NUM,
Collections.emptyList(),
Collections.emptyList(),
customProperties,
false);
assertThatThrownBy(
() ->
admin.createTable(
logTablePathWithoutBucketkeys,
logTableWithoutBucketKeys2,
false)
.get())
.cause()
.isInstanceOf(LakeTableAlreadyExistException.class)
.hasMessage(
"The table `fluss`.`log_table_without_bucket_keys` already exists in Paimon catalog, but the table schema is not compatible. "
+ "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=-1, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, file.format=parquet, fluss.k1=v1}, comment=null}, "
+ "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `c3` STRING, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE], partitionKeys=[], primaryKeys=[], options={bucket=-1, fluss.table.replication.factor=1, fluss.table.datalake.enabled=true, fluss.table.datalake.format=paimon, partition.legacy-name=false, file.format=parquet, fluss.k1=v1}, comment=null}. "
+ "Please first drop the table in Paimon catalog or use a new table name.");
}
```
--
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] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #1847:
URL: https://github.com/apache/fluss/pull/1847
--
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] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1847:
URL: https://github.com/apache/fluss/pull/1847#discussion_r2514016487
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java:
##########
@@ -610,6 +809,84 @@ void testAlterLakeEnabledTableProperties() throws Exception {
admin.alterTable(TablePath.of(DATABASE, "not_exist_table"), tableChanges, true).get();
}
+ @Test
+ void testEnableLakeTableAfterAlterTableProperties() throws Exception {
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("k1", "v1");
+ customProperties.put("paimon.file.format", "parquet");
+
+ // create table
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .build())
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, false)
+ .customProperties(customProperties)
+ .distributedBy(BUCKET_NUM, "c1", "c2")
+ .build();
+ TablePath tablePath = TablePath.of(DATABASE, "enable_lake_table_after_alter_properties");
+ admin.createTable(tablePath, tableDescriptor, false).get();
+ // paimon table should not exist because lake table is disable
+ assertThatThrownBy(
+ () ->
+ paimonCatalog.getTable(
+ Identifier.create(DATABASE, tablePath.getTableName())))
+ .isInstanceOf(Catalog.TableNotExistException.class)
+ .hasMessage(String.format("Table %s does not exist.", tablePath));
+
+ // alter fluss tale properties
+ List<TableChange> tableChanges =
+ Arrays.asList(TableChange.reset("k1"), TableChange.set("k2", "v2"));
+ admin.alterTable(tablePath, tableChanges, false).get();
+ // enable lake table should be ok
+ TableChange.SetOption enableLake =
+ TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
+ admin.alterTable(tablePath, Collections.singletonList(enableLake), false).get();
+ Table paimonTable =
+ paimonCatalog.getTable(Identifier.create(DATABASE, tablePath.getTableName()));
+ customProperties.remove("k1");
+ customProperties.put("k2", "v2");
+ Map<String, String> newProperties = new HashMap<>(tableDescriptor.getProperties());
+ newProperties.put(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
+ tableDescriptor = tableDescriptor.withProperties(newProperties, customProperties);
+ verifyPaimonTable(
+ paimonTable,
+ tableDescriptor,
+ RowType.of(
+ new DataType[] {
+ org.apache.paimon.types.DataTypes.INT(),
+ org.apache.paimon.types.DataTypes.STRING(),
+ // for __bucket, __offset, __timestamp
+ org.apache.paimon.types.DataTypes.INT(),
+ org.apache.paimon.types.DataTypes.BIGINT(),
+ org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()
+ },
+ new String[] {
+ "c1",
+ "c2",
+ BUCKET_COLUMN_NAME,
+ OFFSET_COLUMN_NAME,
+ TIMESTAMP_COLUMN_NAME
+ }),
+ "c1,c2",
+ BUCKET_NUM);
+
+ // disable lake table
+ TableChange.SetOption diableLake =
+ TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false");
+ admin.alterTable(tablePath, Collections.singletonList(diableLake), false).get();
Review Comment:
Corrected spelling of 'diableLake' to 'disableLake'.
```suggestion
TableChange.SetOption disableLake =
TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false");
admin.alterTable(tablePath, Collections.singletonList(disableLake), false).get();
```
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java:
##########
@@ -610,6 +809,84 @@ void testAlterLakeEnabledTableProperties() throws Exception {
admin.alterTable(TablePath.of(DATABASE, "not_exist_table"), tableChanges, true).get();
}
+ @Test
+ void testEnableLakeTableAfterAlterTableProperties() throws Exception {
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("k1", "v1");
+ customProperties.put("paimon.file.format", "parquet");
+
+ // create table
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .build())
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, false)
+ .customProperties(customProperties)
+ .distributedBy(BUCKET_NUM, "c1", "c2")
+ .build();
+ TablePath tablePath = TablePath.of(DATABASE, "enable_lake_table_after_alter_properties");
+ admin.createTable(tablePath, tableDescriptor, false).get();
+ // paimon table should not exist because lake table is disable
+ assertThatThrownBy(
+ () ->
+ paimonCatalog.getTable(
+ Identifier.create(DATABASE, tablePath.getTableName())))
+ .isInstanceOf(Catalog.TableNotExistException.class)
+ .hasMessage(String.format("Table %s does not exist.", tablePath));
+
+ // alter fluss tale properties
Review Comment:
Corrected spelling of 'tale' to 'table' in comment.
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonSchemaValidation.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.paimon.utils;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.exception.TableAlreadyExistException;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.schema.Schema;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.fluss.lake.paimon.utils.PaimonConversions.FLUSS_CONF_PREFIX;
+
+/** Validator of Paimon table schema. */
+public class PaimonSchemaValidation {
+
+ public static void validatePaimonSchemaCapability(Schema existingSchema, Schema newSchema) {
+ // check fields
+ if (!existingSchema.fields().equals(newSchema.fields())) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The fields of the existing Paimon table are not compatible with those of the new table to be created. "
+ + "Existing fields: %s, new fields: %s.",
+ existingSchema.fields(), newSchema.fields()));
+ }
+
+ // check pks
+ if (!existingSchema.primaryKeys().equals(newSchema.primaryKeys())) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The primary keys of the existing Paimon table are not compatible with those of the new table to be created. "
+ + "Existing primary keys: %s, new primary keys: %s.",
+ existingSchema.primaryKeys(), newSchema.primaryKeys()));
+ }
+
+ // check partition keys
+ if (!existingSchema.partitionKeys().equals(newSchema.partitionKeys())) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The partition keys of the existing Paimon table are not compatible with those of the new table to be created. "
+ + "Existing partition keys: %s, new partition keys: %s.",
+ existingSchema.partitionKeys(), newSchema.partitionKeys()));
+ }
+
+ // check options
+ Map<String, String> existingOptions = new HashMap<>(existingSchema.options());
+ Map<String, String> newOptions = new HashMap<>(newSchema.options());
+ // `path` will be set automatically by Paimon, so we need to remove it here
+ existingOptions.remove(CoreOptions.PATH.key());
+ // when enable datalake with an existing table, we should ignore the
+ // `table.datalake.enabled`
+ String datalakeConfigKey = FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_ENABLED.key();
+ if (Boolean.FALSE.toString().equals(existingOptions.get(datalakeConfigKey))) {
+ existingOptions.remove(datalakeConfigKey);
+ newOptions.remove(datalakeConfigKey);
+ }
+ if (!existingOptions.equals(newOptions)) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The options of the existing Paimon table are not compatible with those of the new table to be created. "
+ + "Existing options: %s, new options: %s.",
+ existingOptions, newSchema.options()));
Review Comment:
The error message is inconsistent with the comparison being performed. The code compares `existingOptions` (modified copy) with `newOptions` (modified copy) at line 74, but the error message at line 79 uses `newSchema.options()` (original unmodified options) instead of `newOptions`. This will display incorrect options in the error message. Change `newSchema.options()` to `newOptions`.
```suggestion
existingOptions, newOptions));
```
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java:
##########
@@ -610,6 +809,84 @@ void testAlterLakeEnabledTableProperties() throws Exception {
admin.alterTable(TablePath.of(DATABASE, "not_exist_table"), tableChanges, true).get();
}
+ @Test
+ void testEnableLakeTableAfterAlterTableProperties() throws Exception {
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("k1", "v1");
+ customProperties.put("paimon.file.format", "parquet");
+
+ // create table
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .build())
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, false)
+ .customProperties(customProperties)
+ .distributedBy(BUCKET_NUM, "c1", "c2")
+ .build();
+ TablePath tablePath = TablePath.of(DATABASE, "enable_lake_table_after_alter_properties");
+ admin.createTable(tablePath, tableDescriptor, false).get();
+ // paimon table should not exist because lake table is disable
+ assertThatThrownBy(
+ () ->
+ paimonCatalog.getTable(
+ Identifier.create(DATABASE, tablePath.getTableName())))
+ .isInstanceOf(Catalog.TableNotExistException.class)
+ .hasMessage(String.format("Table %s does not exist.", tablePath));
+
+ // alter fluss tale properties
+ List<TableChange> tableChanges =
+ Arrays.asList(TableChange.reset("k1"), TableChange.set("k2", "v2"));
+ admin.alterTable(tablePath, tableChanges, false).get();
+ // enable lake table should be ok
+ TableChange.SetOption enableLake =
+ TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
+ admin.alterTable(tablePath, Collections.singletonList(enableLake), false).get();
+ Table paimonTable =
+ paimonCatalog.getTable(Identifier.create(DATABASE, tablePath.getTableName()));
+ customProperties.remove("k1");
+ customProperties.put("k2", "v2");
+ Map<String, String> newProperties = new HashMap<>(tableDescriptor.getProperties());
+ newProperties.put(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
+ tableDescriptor = tableDescriptor.withProperties(newProperties, customProperties);
+ verifyPaimonTable(
+ paimonTable,
+ tableDescriptor,
+ RowType.of(
+ new DataType[] {
+ org.apache.paimon.types.DataTypes.INT(),
+ org.apache.paimon.types.DataTypes.STRING(),
+ // for __bucket, __offset, __timestamp
+ org.apache.paimon.types.DataTypes.INT(),
+ org.apache.paimon.types.DataTypes.BIGINT(),
+ org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()
+ },
+ new String[] {
+ "c1",
+ "c2",
+ BUCKET_COLUMN_NAME,
+ OFFSET_COLUMN_NAME,
+ TIMESTAMP_COLUMN_NAME
+ }),
+ "c1,c2",
+ BUCKET_NUM);
+
+ // disable lake table
+ TableChange.SetOption diableLake =
+ TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false");
+ admin.alterTable(tablePath, Collections.singletonList(diableLake), false).get();
+
+ // alter fluss tale properties when lake table is disabled
Review Comment:
Corrected spelling of 'tale' to 'table' in comment.
```suggestion
// alter fluss table properties when lake table is disabled
```
--
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] [server] Tolerate Paimon lake table existent if the schema and properties matches [fluss]
Posted by "LiebingYu (via GitHub)" <gi...@apache.org>.
LiebingYu commented on code in PR #1847:
URL: https://github.com/apache/fluss/pull/1847#discussion_r2516533010
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableManagerITCase.java:
##########
@@ -103,9 +103,7 @@ void testCreateAndGetTable() throws Exception {
lakeTablePath, lakeTableDescriptor, false))
.get())
.cause()
- .isInstanceOf(LakeTableAlreadyExistException.class)
- .hasMessage(
- "The table %s already exists in paimon catalog, please first drop the table in paimon catalog or use a new table name.",
- lakeTablePath);
+ .isInstanceOf(TableAlreadyExistException.class)
+ .hasMessage("Table %s already exists.", lakeTablePath);
Review Comment:
This exception is not related to lake table. It raise because we create a existing fluss table again.
--
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