You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "luoyuxia (via GitHub)" <gi...@apache.org> on 2026/10/12 13:00:55 UTC
[PR] [lake/paimon] Promote existing Paimon log tables to Fluss lake tables [fluss]
luoyuxia opened a new pull request, #4619:
URL: https://github.com/apache/fluss/pull/4619
<!--
*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], [rust], [python], [c++], [elixir]). 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.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [x] Yes (OpenAI Codex, GPT-6)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: OpenAI Codex (GPT-6) following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: N/A. Implements [FIP-25](https://cwiki.apache.org/confluence/spaces/FLUSS/pages/406623572/FIP-25+Support+In-Place+Promotion+of+Existing+Paimon+Tables+to+Datalake-Enabled+Fluss+Tables).
Promote an existing append-only Paimon table to a datalake-enabled Fluss log table in place. Historical data stays in Paimon; promotion registers its current snapshot for Fluss lake reads, without bulk loading it into Fluss.
### Brief change log
- Derive a compatible Fluss log table schema, distribution, and partition configuration from the existing Paimon table, rejecting unsupported semantics.
- Add the create-table-on-lake Admin RPC and validate the existing Paimon snapshot before datalake enablement.
- Add the Flink enable_fluss_on_lake_table procedure to create Fluss metadata, register the initial lake snapshot, and enable lake reads. Repeated calls after successful promotion are idempotent.
- Add unit and integration coverage for metadata conversion, creation, retries, tiering, and union reads.
### Tests
- `./mvnw -pl fluss-lake/fluss-lake-paimon,fluss-flink/fluss-flink-common -am -DskipTests verify` (build, Checkstyle, Spotless, license checks): passed.
- Targeted unit `verify` for LakeStorageTest, TableDescriptorValidationTest, PaimonLakeCatalogTest, PaimonToFlussTableDescriptorConverterTest, EnableFlussOnLakeTableProcedureTest, and ApiManagerTest: passed.
- Targeted integration `verify` for CreateTableOnLakeITCase and EnableFlussOnLakeTableProcedureITCase: 9 tests passed.
- Full `./mvnw verify` was started but stopped at the user's request after the targeted checks passed.
- Known follow-up: retry after Fluss metadata creation but before initial snapshot registration needs explicit recovery coverage.
### API and Format
Adds `Admin.createTableOnLake`, the corresponding RPC (API key 1068), and the Flink procedure. It does not change the Paimon storage format or rewrite historical data. This PR covers append-only Paimon tables; primary-key table promotion and bulk load are separate work. Native Paimon writers and direct schema changes must be stopped before promotion, and writes must go through Fluss afterward.
### Documentation
The behavior is specified in FIP-25 above. This code-only PR leaves user-facing deployment instructions for a follow-up.
### Generative AI disclosure
OpenAI Codex (GPT-6) was used to prepare and adapt this implementation and its tests.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #4619:
URL: https://github.com/apache/fluss/pull/4619#issuecomment-6335150874
@wuchong Thanks for you review. I have address your comment
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #4619:
URL: https://github.com/apache/fluss/pull/4619#discussion_r4266074660
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/EnableFlussOnLakeTableProcedure.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.TableAlreadyExistException;
+import org.apache.fluss.flink.tiering.committer.FlussTableLakeSnapshotCommitter;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.lake.lakestorage.LakeCatalog;
+import org.apache.fluss.lake.lakestorage.LakeStorage;
+import org.apache.fluss.lake.lakestorage.LakeStoragePlugin;
+import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.utils.ExceptionUtils;
+import org.apache.fluss.utils.TemporaryClassLoaderContext;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
+
+/** Procedure for promoting an existing Paimon log table to a datalake-enabled Fluss table. */
+public class EnableFlussOnLakeTableProcedure extends ProcedureBase {
+
+ private static final String SUCCESS_MESSAGE =
+ "Successfully enabled Fluss on Paimon table '%s'.";
+
+ @ProcedureHint(argument = {@ArgumentHint(name = "table", type = @DataTypeHint("STRING"))})
+ public String[] call(ProcedureContext context, String table) throws Exception {
+ return enable(table, Collections.<String, String>emptyMap());
+ }
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "options", type = @DataTypeHint("STRING"))
+ })
+ public String[] call(ProcedureContext context, String table, String options) throws Exception {
+ return enable(table, parseOptions(options));
+ }
+
+ private String[] enable(String table, Map<String, String> options) throws Exception {
+ TablePath tablePath = parseTablePath(table, flussProcedureContext.getDefaultDatabase());
+ TableInfo tableInfo;
+ boolean created = false;
+ try {
+ tableInfo = admin.createTableOnLake(tablePath, options).get();
+ created = true;
+ } catch (ExecutionException e) {
+ if (!ExceptionUtils.findThrowable(e, TableAlreadyExistException.class).isPresent()) {
+ throw e;
+ }
+ tableInfo = admin.getTableInfo(tablePath).get();
+ }
+
+ if (created) {
+ registerInitialSnapshot(tableInfo);
Review Comment:
Update after checking the append-only log-table read path: my previous reply was too strict about proving that no Fluss writes occurred. `commitInitialSnapshot` records an empty Fluss offset map. For a bucket without a snapshot offset, lake/log reads start from the earliest Fluss log offset, so initial registration does not skip Fluss log records written before registration.
For an existing table that is still datalake-disabled, the procedure can query `Admin.getLatestLakeSnapshot`: if a snapshot is already registered, leave it intact; if it reports `LakeTableSnapshotNotExistException`, register the initial Paimon data-change snapshot and continue with enablement. The existing alter path still checks schema compatibility and equality with Paimon's latest data-change snapshot. I have added this retry path and a test for the failure window locally; the branch has not been pushed yet.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #4619:
URL: https://github.com/apache/fluss/pull/4619#discussion_r4346967198
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java:
##########
@@ -152,31 +162,51 @@ private static boolean equalPhysicalSchema(Schema existingSchema, Schema newSche
private static boolean equalPaimonOptions(
Map<String, String> existingOptions, Map<String, String> newOptions) {
- if (!Objects.equals(
- existingOptions.get(PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY),
- Boolean.FALSE.toString())) {
- return false;
- }
-
+ Options existing = Options.fromMap(existingOptions);
+ Options proposed = Options.fromMap(newOptions);
for (String option : PAIMON_UNSETTABLE_OPTIONS) {
- if (!Objects.equals(existingOptions.get(option), newOptions.get(option))) {
+ if (PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY.equals(option)) {
+ // Existing Paimon tables keep their partition path encoding. Fluss addresses
+ // partitions through logical values and Paimon APIs rather than the directory name.
+ continue;
+ }
+ if (!equalPaimonOption(option, existing, proposed)) {
return false;
}
}
for (String option : CoreOptions.IMMUTABLE_OPTIONS) {
- if (!Objects.equals(existingOptions.get(option), newOptions.get(option))) {
+ if (!equalPaimonOption(option, existing, proposed)) {
return false;
}
}
return true;
}
+ private static boolean equalPaimonOption(String key, Options existing, Options proposed) {
+ ConfigOption<?> option = PAIMON_CORE_OPTIONS.get(key);
+ if (option != null) {
+ return Objects.equals(existing.get(option), proposed.get(option));
+ }
+ return Objects.equals(existing.get(key), proposed.get(key));
+ }
+
private static boolean equalDataFieldIgnoreFieldId(
DataField existingField, DataField newField) {
- return Objects.equals(existingField.name(), newField.name())
- && existingField.type().equalsIgnoreFieldId(newField.type());
+ if (!Objects.equals(existingField.name(), newField.name())) {
+ return false;
+ }
+
+ DataType existingType = existingField.type();
+ // Fluss maps VARCHAR(n) to STRING and VARBINARY(n) to BYTES. Converting the Fluss
+ // descriptor back to Paimon yields unbounded types, so ignore length but keep nullability.
+ if (existingType instanceof VarCharType) {
+ existingType = DataTypes.STRING().copy(existingType.isNullable());
+ } else if (existingType instanceof VarBinaryType) {
+ existingType = DataTypes.BYTES().copy(existingType.isNullable());
+ }
+ return existingType.equalsIgnoreFieldId(newField.type());
Review Comment:
Could we apply this normalization recursively to ARRAY elements, MAP keys/values, and ROW fields? `PaimonDataTypeToFlussDataType` recursively converts nested `VARCHAR(n)`/`VARBINARY(n)` to `STRING`/`BYTES`, but this check only normalizes the outer field type. For example, `ARRAY<VARCHAR(20)>` is still compared with `ARRAY<STRING>` and rejected.
I reproduced this through the actual procedure with `ARRAY<VARCHAR(20) NOT NULL>`, `ARRAY<VARBINARY(128)>`, `MAP<VARCHAR(20), VARBINARY(128)>`, and `ROW<name VARCHAR(20)>`. In all four cases, Fluss metadata is created, enablement fails with `LakeTableAlreadyExistException`, and retrying fails again.
A `DataTypeVisitor` could handle recursive type replacement or compatibility comparison while preserving nullability at every nesting level and the other schema checks. Please also add procedure-level regression tests for these nested mappings, including nullability-mismatch rejection.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #4619:
URL: https://github.com/apache/fluss/pull/4619#discussion_r4339257498
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -308,6 +352,76 @@ private void createTable(TablePath tablePath, Schema schema, boolean isCreatingF
}
}
+ private static void validateSnapshotForEnablement(
+ TablePath tablePath,
+ FileStoreTable fileStoreTable,
+ @Nullable Long registeredLakeSnapshotId) {
+ SnapshotManager snapshotManager = fileStoreTable.snapshotManager();
+ Long latestPaimonSnapshotId = snapshotManager.latestSnapshotId();
+ if (Objects.equals(registeredLakeSnapshotId, latestPaimonSnapshotId)) {
Review Comment:
The registered snapshot itself is not checked here. If snapshot 5 has expired and snapshot 6 is only a COMPACT, the later-snapshot loop passes, but Fluss keeps snapshot ID 5 and `PaimonSplitPlanner` fails when reading it.
Please check that a non-null registered snapshot still exists **before the equal-ID early return**, so that branch cannot bypass the check either. Both IDs being null should remain valid. Please also update the expired-snapshot test to expect enablement to fail, and revise the comment below that currently allows an expired registered snapshot.
```suggestion
if (registeredLakeSnapshotId != null
&& !snapshotManager.snapshotExists(registeredLakeSnapshotId)) {
throw invalidSnapshotForEnablement(
tablePath,
registeredLakeSnapshotId,
latestPaimonSnapshotId,
"The registered Paimon snapshot has expired");
}
if (Objects.equals(registeredLakeSnapshotId, latestPaimonSnapshotId)) {
```
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java:
##########
@@ -292,6 +302,50 @@ public static List<SchemaChange> toPaimonSchemaChanges(
return schemaChanges;
}
+ // ------------------------------------------------------------------------
+ // Table metadata conversions
+ // ------------------------------------------------------------------------
+
+ /** Converts an existing Paimon append-only table to a Fluss table descriptor. */
+ public static TableDescriptor toFlussTableDescriptor(
+ TablePath tablePath, FileStoreTable table) {
+ if (!table.primaryKeys().isEmpty()) {
+ throw new InvalidTableException(
+ String.format(
+ "Converting the Paimon primary-key table %s to a Fluss table is not supported yet.",
+ tablePath));
+ }
+ validateAppendOnlyTable(tablePath, table);
+
+ List<DataField> dataFields = table.schema().fields();
+ org.apache.fluss.metadata.Schema.Builder schemaBuilder =
+ org.apache.fluss.metadata.Schema.newBuilder();
+ for (DataField dataField : dataFields) {
+ try {
+ schemaBuilder
+ .column(
+ dataField.name(),
+ dataField.type().accept(PaimonDataTypeToFlussDataType.INSTANCE))
Review Comment:
Could we relax `isPaimonSchemaCompatible` to accept the `VARCHAR(n)`/`STRING` and `VARBINARY(n)`/`BYTES` mappings used here? For example, `VARCHAR(20)` and `VARBINARY(128)` are converted to Fluss `STRING` and `BYTES`, but converting the descriptor back to Paimon during enablement produces unbounded types. The exact type comparison then rejects the existing lake table after the Fluss table has already been created, and retrying the procedure fails again. Please allow these two mappings while preserving nullability and the other schema checks, and add procedure-level regression tests that complete enablement for both bounded types.
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java:
##########
@@ -152,13 +152,12 @@ private static boolean equalPhysicalSchema(Schema existingSchema, Schema newSche
private static boolean equalPaimonOptions(
Map<String, String> existingOptions, Map<String, String> newOptions) {
- if (!Objects.equals(
- existingOptions.get(PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY),
- Boolean.FALSE.toString())) {
- return false;
- }
-
for (String option : PAIMON_UNSETTABLE_OPTIONS) {
+ if (PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY.equals(option)) {
+ // Existing Paimon tables keep their partition path encoding. Fluss addresses
+ // partitions through logical values and Paimon APIs rather than the directory name.
+ continue;
+ }
if (!Objects.equals(existingOptions.get(option), newOptions.get(option))) {
Review Comment:
`equalPaimonOptions` compares raw map entries, so two supported append-only configurations are rejected during enablement:
1. An append-only table with no `bucket` option has an effective bucket count of `-1` (`BUCKET_UNAWARE`). Converting it to a Fluss descriptor and then back to a Paimon schema explicitly writes `bucket=-1`, so this check compares `null` with `"-1"`.
2. A table with `bucket=2`, `bucket-key=id`, and `bucket-function.type=default` is `HASH_FIXED` and passes `applyDistribution`. However, the regenerated schema omits the default bucket function, so the immutable-option check compares `"default"` with `null`.
Both cases reproduce through the real `PaimonLakeCatalog.createTable` enablement path: it throws `TableAlreadyExistException` reporting an incompatible schema, even though the parsed bucket count and bucket function are unchanged. The corresponding controls (explicit `bucket=-1`, and omitted `bucket-function.type` for the fixed-bucket table) pass.
Could we compare the parsed effective configuration values, including defaults (e.g. `CoreOptions.bucket()` and `bucketFunctionType()`), while still rejecting genuinely different values and preserving the other schema checks? Please add enablement regression tests for both cases and their controls.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #4619:
URL: https://github.com/apache/fluss/pull/4619#discussion_r4266074660
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/EnableFlussOnLakeTableProcedure.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.TableAlreadyExistException;
+import org.apache.fluss.flink.tiering.committer.FlussTableLakeSnapshotCommitter;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.lake.lakestorage.LakeCatalog;
+import org.apache.fluss.lake.lakestorage.LakeStorage;
+import org.apache.fluss.lake.lakestorage.LakeStoragePlugin;
+import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.utils.ExceptionUtils;
+import org.apache.fluss.utils.TemporaryClassLoaderContext;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
+
+/** Procedure for promoting an existing Paimon log table to a datalake-enabled Fluss table. */
+public class EnableFlussOnLakeTableProcedure extends ProcedureBase {
+
+ private static final String SUCCESS_MESSAGE =
+ "Successfully enabled Fluss on Paimon table '%s'.";
+
+ @ProcedureHint(argument = {@ArgumentHint(name = "table", type = @DataTypeHint("STRING"))})
+ public String[] call(ProcedureContext context, String table) throws Exception {
+ return enable(table, Collections.<String, String>emptyMap());
+ }
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "options", type = @DataTypeHint("STRING"))
+ })
+ public String[] call(ProcedureContext context, String table, String options) throws Exception {
+ return enable(table, parseOptions(options));
+ }
+
+ private String[] enable(String table, Map<String, String> options) throws Exception {
+ TablePath tablePath = parseTablePath(table, flussProcedureContext.getDefaultDatabase());
+ TableInfo tableInfo;
+ boolean created = false;
+ try {
+ tableInfo = admin.createTableOnLake(tablePath, options).get();
+ created = true;
+ } catch (ExecutionException e) {
+ if (!ExceptionUtils.findThrowable(e, TableAlreadyExistException.class).isPresent()) {
+ throw e;
+ }
+ tableInfo = admin.getTableInfo(tablePath).get();
+ }
+
+ if (created) {
+ registerInitialSnapshot(tableInfo);
Review Comment:
Correction to my earlier reply: `getLatestLakeSnapshot` only tells us whether a lake snapshot is registered. A datalake-disabled table with no registered snapshot is not necessarily an abandoned promotion, and those facts do not prove that no Fluss writes occurred. Checking offsets would not be atomic with concurrent writes either.
So the existing read API alone is insufficient for safe automatic repair. That would require persistent promotion state tied to the table ID plus a write fence during initialization, or an equivalent atomic guarantee. With the current implementation, the recovery for a table left behind before snapshot registration is to drop that incomplete Fluss table and rerun the procedure. I retract the suggestion that we can simply re-register based on the snapshot query.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #4619:
URL: https://github.com/apache/fluss/pull/4619#discussion_r4266058317
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/EnableFlussOnLakeTableProcedure.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.TableAlreadyExistException;
+import org.apache.fluss.flink.tiering.committer.FlussTableLakeSnapshotCommitter;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.lake.lakestorage.LakeCatalog;
+import org.apache.fluss.lake.lakestorage.LakeStorage;
+import org.apache.fluss.lake.lakestorage.LakeStoragePlugin;
+import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.utils.ExceptionUtils;
+import org.apache.fluss.utils.TemporaryClassLoaderContext;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
+
+/** Procedure for promoting an existing Paimon log table to a datalake-enabled Fluss table. */
+public class EnableFlussOnLakeTableProcedure extends ProcedureBase {
+
+ private static final String SUCCESS_MESSAGE =
+ "Successfully enabled Fluss on Paimon table '%s'.";
+
+ @ProcedureHint(argument = {@ArgumentHint(name = "table", type = @DataTypeHint("STRING"))})
+ public String[] call(ProcedureContext context, String table) throws Exception {
+ return enable(table, Collections.<String, String>emptyMap());
+ }
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "options", type = @DataTypeHint("STRING"))
+ })
+ public String[] call(ProcedureContext context, String table, String options) throws Exception {
+ return enable(table, parseOptions(options));
+ }
+
+ private String[] enable(String table, Map<String, String> options) throws Exception {
+ TablePath tablePath = parseTablePath(table, flussProcedureContext.getDefaultDatabase());
+ TableInfo tableInfo;
+ boolean created = false;
+ try {
+ tableInfo = admin.createTableOnLake(tablePath, options).get();
+ created = true;
+ } catch (ExecutionException e) {
+ if (!ExceptionUtils.findThrowable(e, TableAlreadyExistException.class).isPresent()) {
+ throw e;
+ }
+ tableInfo = admin.getTableInfo(tablePath).get();
+ }
+
+ if (created) {
+ registerInitialSnapshot(tableInfo);
Review Comment:
Good catch. This retry gap is real. `Admin.getLatestLakeSnapshot` already gives us the read we need: it fails with `LakeTableSnapshotNotExistException` when Fluss has not registered a snapshot, so a new RPC is unnecessary.
For an existing Fluss table that is still datalake-disabled, we can distinguish a completed registration from a missing one. If a snapshot is already registered, the procedure should skip registration and continue to the existing Paimon/Fluss snapshot equality check. If none is registered, it can retry initial registration only after establishing that the table is still in the promotion initialization state and has not accepted Fluss writes. An unsafe or uncertain state should fail rather than silently resetting the handoff. I'll address this path and add a retry test for failure between table creation and snapshot registration.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #4619:
URL: https://github.com/apache/fluss/pull/4619#issuecomment-6339989929
Let me merge this first. I'll address the remain one comment in the follow pr.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4619:
URL: https://github.com/apache/fluss/pull/4619#discussion_r4259669672
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/EnableFlussOnLakeTableProcedure.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.TableAlreadyExistException;
+import org.apache.fluss.flink.tiering.committer.FlussTableLakeSnapshotCommitter;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.lake.lakestorage.LakeCatalog;
+import org.apache.fluss.lake.lakestorage.LakeStorage;
+import org.apache.fluss.lake.lakestorage.LakeStoragePlugin;
+import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.utils.ExceptionUtils;
+import org.apache.fluss.utils.TemporaryClassLoaderContext;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
+
+/** Procedure for promoting an existing Paimon log table to a datalake-enabled Fluss table. */
+public class EnableFlussOnLakeTableProcedure extends ProcedureBase {
+
+ private static final String SUCCESS_MESSAGE =
+ "Successfully enabled Fluss on Paimon table '%s'.";
+
+ @ProcedureHint(argument = {@ArgumentHint(name = "table", type = @DataTypeHint("STRING"))})
+ public String[] call(ProcedureContext context, String table) throws Exception {
+ return enable(table, Collections.<String, String>emptyMap());
+ }
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "options", type = @DataTypeHint("STRING"))
+ })
+ public String[] call(ProcedureContext context, String table, String options) throws Exception {
+ return enable(table, parseOptions(options));
+ }
+
+ private String[] enable(String table, Map<String, String> options) throws Exception {
+ TablePath tablePath = parseTablePath(table, flussProcedureContext.getDefaultDatabase());
+ TableInfo tableInfo;
+ boolean created = false;
+ try {
+ tableInfo = admin.createTableOnLake(tablePath, options).get();
+ created = true;
+ } catch (ExecutionException e) {
+ if (!ExceptionUtils.findThrowable(e, TableAlreadyExistException.class).isPresent()) {
+ throw e;
+ }
+ tableInfo = admin.getTableInfo(tablePath).get();
+ }
+
+ if (created) {
+ registerInitialSnapshot(tableInfo);
Review Comment:
The procedure only registers the initial lake snapshot when it *just created* Fluss metadata. If a previous attempt created the table but failed before snapshot registration, subsequent calls hit `TableAlreadyExistException`, skip snapshot registration, and will likely fail the enablement due to the Fluss-vs-Paimon snapshot mismatch check. To make retries operationally robust, consider adding logic to detect/repair the “created but snapshot not registered” state (e.g., by querying the registered snapshot via an admin/RPC read, or by making snapshot registration idempotent under a clearly safe condition) before attempting `alterTable(...TABLE_DATALAKE_ENABLED=true)`.
##########
fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java:
##########
@@ -1116,6 +1147,38 @@ private static void handleListOffsetsResponse(
}
}
+ private static TableInfo toTableInfo(TablePath tablePath, CreateTableOnLakeResponse response) {
+ return toTableInfo(
+ tablePath,
+ response.getTableId(),
+ response.getSchemaId(),
+ response.getTableJson(),
+ response.hasRemoteDataDir() ? response.getRemoteDataDir() : null,
+ response.getCreatedTime(),
+ response.getModifiedTime(),
+ 0L);
+ }
Review Comment:
`CreateTableFromLakeResponse` doesn’t carry `bucketCountEpoch`, so the client constructs `TableInfo` with a hard-coded `0L`. If `bucketCountEpoch` is used for correctness (e.g., detecting bucket-count changes), this creates an inconsistency between `createTableFromLake` and `getTableInfo`. Consider adding `bucket_count_epoch` to `PbTableInfo` (and populating it server-side) or returning only identifiers and requiring clients to call `getTableInfo` for a fully populated `TableInfo`.
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java:
##########
@@ -96,19 +100,43 @@ protected Catalog getPaimonCatalog() {
return paimonCatalog;
}
+ @Override
+ public TableDescriptor getTableDescriptor(TablePath tablePath) throws TableNotExistException {
+ try {
+ Table table = paimonCatalog.getTable(toPaimon(tablePath));
+ if (!(table instanceof FileStoreTable)) {
+ throw new InvalidTableException(
+ String.format("Paimon table %s is not a file store table.", tablePath));
+ }
+ return PaimonToFlussTableDescriptorConverter.convert(tablePath, (FileStoreTable) table);
+ } catch (Catalog.TableNotExistException e) {
+ throw new TableNotExistException("Table " + tablePath + " does not exist in Paimon.");
+ }
+ }
+
+ @Override
+ public Optional<Long> getLatestSnapshotId(TablePath tablePath) throws TableNotExistException {
+ try {
+ Table table = paimonCatalog.getTable(toPaimon(tablePath));
+ return getLatestSnapshotId((FileStoreTable) table);
Review Comment:
`getLatestDataChangeSnapshotId(TablePath)` blindly casts to `FileStoreTable`, which can throw a `ClassCastException` at runtime for non-FileStore Paimon tables. Align this with `getTableDescriptor(...)` by checking `instanceof FileStoreTable` and throwing a domain exception (e.g., `InvalidTableException`) with a clear message when the table type is unsupported.
--
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] Promote existing Paimon log tables to Fluss lake tables [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #4619:
URL: https://github.com/apache/fluss/pull/4619
--
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