This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 89b7c0cad [lake/paimon] Support custom Paimon path (#3523)
89b7c0cad is described below
commit 89b7c0cad6b6c3e7e931fb1848a0cf597c89481d
Author: Liurnly <[email protected]>
AuthorDate: Mon Jul 13 21:56:43 2026 +0800
[lake/paimon] Support custom Paimon path (#3523)
---
.../fluss/lake/paimon/PaimonLakeCatalog.java | 50 +++++++-
.../fluss/lake/paimon/utils/PaimonConversions.java | 1 -
.../lake/paimon/LakeEnabledTableCreateITCase.java | 136 +++++++++++++++++++++
3 files changed, 184 insertions(+), 3 deletions(-)
diff --git
a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java
b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java
index 738f659dd..bd846cec2 100644
---
a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java
+++
b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java
@@ -32,6 +32,7 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
@@ -42,6 +43,7 @@ import org.apache.paimon.types.DataTypes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.stream.Collectors;
@@ -59,6 +61,7 @@ import static
org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
public class PaimonLakeCatalog implements LakeCatalog {
private static final Logger LOG =
LoggerFactory.getLogger(PaimonLakeCatalog.class);
+ private static final String PAIMON_PATH_KEY = "paimon.path";
public static final LinkedHashMap<String, DataType> SYSTEM_COLUMNS = new
LinkedHashMap<>();
static {
@@ -115,6 +118,14 @@ public class PaimonLakeCatalog implements LakeCatalog {
try {
Table table = paimonCatalog.getTable(toPaimon(tablePath));
FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<TableChange> changesToApply =
+
validateAndFilterPaimonPathChanges(fileStoreTable.location(), tableChanges);
+
+ // Avoid creating a new Paimon schema version for a path-only
no-op.
+ if (changesToApply.isEmpty()) {
+ return;
+ }
+
Schema currentPaimonSchema = fileStoreTable.schema().toSchema();
List<SchemaChange> paimonSchemaChanges;
@@ -122,13 +133,13 @@ public class PaimonLakeCatalog implements LakeCatalog {
currentPaimonSchema,
toPaimonSchema(context.getCurrentTable()))) {
// if the paimon schema is same as current fluss schema,
directly apply all the
// changes.
- paimonSchemaChanges = toPaimonSchemaChanges(tableChanges);
+ paimonSchemaChanges = toPaimonSchemaChanges(changesToApply);
} else if (isPaimonSchemaCompatible(
currentPaimonSchema,
toPaimonSchema(context.getExpectedTable()))) {
// if the schema is same as applied fluss schema , skip adding
columns.
paimonSchemaChanges =
toPaimonSchemaChanges(
- tableChanges.stream()
+ changesToApply.stream()
.filter(
tableChange ->
!(tableChange
@@ -157,6 +168,41 @@ public class PaimonLakeCatalog implements LakeCatalog {
}
}
+ private static List<TableChange> validateAndFilterPaimonPathChanges(
+ Path currentLocation, List<TableChange> tableChanges) {
+ List<TableChange> changesToApply = new
ArrayList<>(tableChanges.size());
+
+ // `paimon.path` is create-time-only because changing it does not
migrate existing data.
+ // It may be repeated after table creation when enabling lake in the
same ALTER. Treat an
+ // equivalent path as a no-op, but reject any actual path change.
+ for (TableChange tableChange : tableChanges) {
+ if (tableChange instanceof TableChange.SetOption) {
+ TableChange.SetOption setOption = (TableChange.SetOption)
tableChange;
+ if (PAIMON_PATH_KEY.equals(setOption.getKey())) {
+ if (currentLocation.equals(new
Path(setOption.getValue()))) {
+ continue;
+ }
+ throw invalidPaimonPathChangeException();
+ }
+ } else if (tableChange instanceof TableChange.ResetOption) {
+ TableChange.ResetOption resetOption =
(TableChange.ResetOption) tableChange;
+ if (PAIMON_PATH_KEY.equals(resetOption.getKey())) {
+ throw invalidPaimonPathChangeException();
+ }
+ }
+ changesToApply.add(tableChange);
+ }
+
+ return changesToApply;
+ }
+
+ private static InvalidAlterTableException
invalidPaimonPathChangeException() {
+ return new InvalidAlterTableException(
+ String.format(
+ "'%s' can only be altered before the Paimon table is
created.",
+ PAIMON_PATH_KEY));
+ }
+
private void createTable(TablePath tablePath, Schema schema, boolean
isCreatingFlussTable)
throws Catalog.DatabaseNotExistException {
Identifier paimonPath = toPaimon(tablePath);
diff --git
a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
index 4bb12a0f4..2aa1f0529 100644
---
a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
+++
b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
@@ -72,7 +72,6 @@ public class PaimonConversions {
static {
PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.BUCKET.key());
PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.BUCKET_KEY.key());
- PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.PATH.key());
PAIMON_UNSETTABLE_OPTIONS.add(PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY);
}
diff --git
a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
index 0efa057b8..52087992a 100644
---
a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
+++
b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
@@ -23,6 +23,7 @@ import org.apache.fluss.client.admin.Admin;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidConfigException;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.LakeTableAlreadyExistException;
@@ -46,6 +47,7 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
@@ -390,6 +392,8 @@ class LakeEnabledTableCreateITCase {
@Test
void testCreateLakeEnableTableWithUnsettablePaimonOptions() {
+
assertThat(PAIMON_UNSETTABLE_OPTIONS).doesNotContain(CoreOptions.PATH.key());
+
Map<String, String> customProperties = new HashMap<>();
for (String key : PAIMON_UNSETTABLE_OPTIONS) {
@@ -418,6 +422,138 @@ class LakeEnabledTableCreateITCase {
}
}
+ @Test
+ void testAlterPaimonPathOnlyWhenPaimonTableNotCreated() throws Exception {
+ TableDescriptor lakeDisabledTable =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .build())
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, false)
+ .distributedBy(BUCKET_NUM, "c1", "c2")
+ .build();
+ TablePath lakeDisabledTablePath = TablePath.of(DATABASE,
"alter_paimon_path_disabled");
+ admin.createTable(lakeDisabledTablePath, lakeDisabledTable,
false).get();
+
+ String customPaimonPath =
+
Files.createTempDirectory("alter-paimon-path-disabled").toUri().toString();
+ admin.alterTable(
+ lakeDisabledTablePath,
+
Collections.singletonList(TableChange.set("paimon.path", customPaimonPath)),
+ false)
+ .get();
+ assertThat(
+ admin.getTableInfo(lakeDisabledTablePath)
+ .get()
+ .toTableDescriptor()
+ .getCustomProperties())
+ .containsEntry("paimon.path", customPaimonPath);
+ assertThatThrownBy(
+ () ->
+ paimonCatalog.getTable(
+ Identifier.create(
+ DATABASE,
lakeDisabledTablePath.getTableName())))
+ .isInstanceOf(Catalog.TableNotExistException.class)
+ .hasMessageContaining(
+ String.format(
+ "Table %s.%s does not exist",
+ DATABASE,
lakeDisabledTablePath.getTableName()));
+
+ TableDescriptor lakeEnabledTable =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .build())
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .distributedBy(BUCKET_NUM, "c1", "c2")
+ .build();
+ TablePath lakeEnabledTablePath = TablePath.of(DATABASE,
"alter_paimon_path_enabled");
+ admin.createTable(lakeEnabledTablePath, lakeEnabledTable, false).get();
+
+ Identifier lakeEnabledPaimonPath =
+ Identifier.create(DATABASE,
lakeEnabledTablePath.getTableName());
+ FileStoreTable paimonTable = (FileStoreTable)
paimonCatalog.getTable(lakeEnabledPaimonPath);
+ String paimonPath = paimonTable.location().toString();
+
+ admin.alterTable(
+ lakeEnabledTablePath,
+ Arrays.asList(
+ TableChange.set("paimon.path", paimonPath +
"/"),
+ TableChange.set("key", "value")),
+ false)
+ .get();
+
+ paimonTable = (FileStoreTable)
paimonCatalog.getTable(lakeEnabledPaimonPath);
+ assertThat(paimonTable.location().toString()).isEqualTo(paimonPath);
+ assertThat(paimonTable.options()).containsEntry("fluss.key", "value");
+ long schemaId = paimonTable.schema().id();
+
+ admin.alterTable(
+ lakeEnabledTablePath,
+
Collections.singletonList(TableChange.set("paimon.path", paimonPath)),
+ false)
+ .get();
+
+ paimonTable = (FileStoreTable)
paimonCatalog.getTable(lakeEnabledPaimonPath);
+ assertThat(paimonTable.location().toString()).isEqualTo(paimonPath);
+ assertThat(paimonTable.schema().id()).isEqualTo(schemaId);
+
+ assertThatThrownBy(
+ () ->
+ admin.alterTable(
+ lakeEnabledTablePath,
+ Collections.singletonList(
+ TableChange.set(
+ "paimon.path",
+
Files.createTempDirectory(
+
"alter-paimon-path-enabled")
+
.toUri()
+
.toString())),
+ false)
+ .get())
+ .cause()
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining(
+ "'paimon.path' can only be altered before the Paimon
table is created");
+
+ TablePath lakeReDisabledTablePath =
+ TablePath.of(DATABASE,
"alter_paimon_path_created_then_disabled");
+ admin.createTable(lakeReDisabledTablePath, lakeEnabledTable,
false).get();
+ Identifier lakeReDisabledPaimonPath =
+ Identifier.create(DATABASE,
lakeReDisabledTablePath.getTableName());
+ paimonCatalog.getTable(lakeReDisabledPaimonPath);
+
+ admin.alterTable(
+ lakeReDisabledTablePath,
+ Collections.singletonList(
+ TableChange.set(
+
ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false")),
+ false)
+ .get();
+
+ assertThatThrownBy(
+ () ->
+ admin.alterTable(
+ lakeReDisabledTablePath,
+ Collections.singletonList(
+ TableChange.set(
+ "paimon.path",
+
Files.createTempDirectory(
+
"alter-paimon-path-created")
+
.toUri()
+
.toString())),
+ false)
+ .get())
+ .cause()
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining(
+ "'paimon.path' can only be altered before the Paimon
table is created");
+ }
+
@Test
void testCreateLakeEnableTableWithExistLakeTable() throws Exception {
Map<String, String> customProperties = new HashMap<>();
