Copilot commented on code in PR #3476:
URL: https://github.com/apache/fluss/pull/3476#discussion_r3400622515
##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkTable.scala:
##########
@@ -68,6 +68,10 @@ class SparkTable(
flussConfig.get(SparkFlussConf.SCAN_START_UP_MODE))
.toUpperCase
val isFullMode = startupMode == SparkFlussConf.StartUpMode.FULL.toString
+ if (isDataLakeEnabled &&
LakeTableUtil.hasCustomLakePath(tableInfo.getProperties)) {
+ throw new UnsupportedOperationException(
+ "Custom lake table path is not supported in Spark connector yet.")
+ }
Review Comment:
This currently rejects *any* Spark access when `table.datalake.enabled=true`
and a custom lake path mapping is set, even for scan modes that may not touch
the lake (e.g., non-FULL modes). The PR description says only Spark *lake
reads* are rejected; consider narrowing the guard to the code paths that
actually read from the lake (e.g., FULL mode / `$lake`-equivalent paths), or
adjust the behavior/message to match the intended restriction.
##########
fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java:
##########
@@ -1365,6 +1365,16 @@ public Optional<LakeTable> getLakeTable(long tableId)
throws Exception {
return getOrEmpty(zkPath).map(LakeTableZNode::decode);
}
+ /** Deletes the {@link LakeTable} for the given table ID if it exists. */
+ public void deleteLakeTable(long tableId) throws Exception {
+ String zkPath = LakeTableZNode.path(tableId);
+ try {
+ zkClient.delete().forPath(zkPath);
+ } catch (KeeperException.NoNodeException ignored) {
+ // Ignore if the lake table progress has not been committed yet.
+ }
+ }
Review Comment:
Using `delete().forPath(...)` will fail if the lake-table znode ever gains
children (now or in future schema evolutions). To make this deletion more
robust, consider using Curator’s `deletingChildrenIfNeeded()` (and optionally
`guaranteed()`) so cleanup does not unexpectedly fail due to node structure
changes.
##########
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java:
##########
@@ -79,6 +79,24 @@ public void registerLakeTableSnapshotV2(
registerLakeTableSnapshotV2(tableId, lakeSnapshotMetadata, null);
}
+ /** Clears the committed lake table progress for the given table. */
+ public void clearLakeTableProgress(long tableId) throws Exception {
+ Optional<LakeTable> optLakeTable = zkClient.getLakeTable(tableId);
+ if (!optLakeTable.isPresent()) {
+ return;
+ }
+
+ zkClient.deleteLakeTable(tableId);
+ LakeTable lakeTable = optLakeTable.get();
+ List<LakeTable.LakeSnapshotMetadata> lakeSnapshotMetadatas =
+ lakeTable.getLakeSnapshotMetadatas();
+ if (lakeSnapshotMetadatas != null) {
+ for (LakeTable.LakeSnapshotMetadata lakeSnapshotMetadata :
lakeSnapshotMetadatas) {
+ lakeSnapshotMetadata.discard();
+ }
+ }
+ }
Review Comment:
The ZK progress node is deleted before discarding snapshot metadata. If
`discard()` throws (e.g., IO failure deleting offset files), the cluster loses
the reference to the progress but may leak local/remote files. Consider
discarding first and deleting the ZK node after successful cleanup, or use a
try/finally pattern that attempts cleanup for all snapshots and only then
deletes (or records) progress removal so failures don’t leave orphaned data.
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateWithHiveCatalogITCase.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * 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;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+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.metadata.Schema;
+import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.types.DataTypes;
+
+import org.apache.paimon.catalog.AbstractCatalog;
+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.options.Options;
+import org.apache.paimon.table.Table;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.fluss.server.utils.LakeStorageUtils.extractLakeProperties;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for create lake enabled table with Paimon Hive catalog. */
+class LakeEnabledTableCreateWithHiveCatalogITCase {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(3)
+ .setClusterConf(initConfig())
+ .build();
+
+ private static final String DATABASE = "fluss";
+ private static final int BUCKET_NUM = 3;
+
+ private static Catalog paimonCatalog;
Review Comment:
The test creates a static `paimonCatalog` but never closes it. Some catalog
implementations keep threads / open resources, which can lead to flaky builds
or hanging test JVMs. Add an `@AfterAll` (or equivalent) to close
`paimonCatalog` when the test class finishes to ensure deterministic cleanup.
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateWithHiveCatalogITCase.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * 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;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+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.metadata.Schema;
+import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.types.DataTypes;
+
+import org.apache.paimon.catalog.AbstractCatalog;
+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.options.Options;
+import org.apache.paimon.table.Table;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.fluss.server.utils.LakeStorageUtils.extractLakeProperties;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for create lake enabled table with Paimon Hive catalog. */
+class LakeEnabledTableCreateWithHiveCatalogITCase {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(3)
+ .setClusterConf(initConfig())
+ .build();
+
+ private static final String DATABASE = "fluss";
+ private static final int BUCKET_NUM = 3;
+
+ private static Catalog paimonCatalog;
+ private static String warehousePath;
+ private static String customTablePath;
+
+ private Connection conn;
+ private Admin admin;
+
+ @BeforeEach
+ protected void setup() {
+ conn =
ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ admin = conn.getAdmin();
+ }
+
+ @AfterEach
+ protected void teardown() throws Exception {
+ if (admin != null) {
+ admin.close();
+ admin = null;
+ }
+
+ if (conn != null) {
+ conn.close();
+ conn = null;
+ }
+ }
+
+ private static Configuration initConfig() {
+ Configuration conf = new Configuration();
+ conf.setString("datalake.format", "paimon");
+ conf.setString("datalake.paimon.metastore", "hive");
+ conf.setString("datalake.paimon.cache-enabled", "false");
+ try {
+ java.nio.file.Path baseDir =
Files.createTempDirectory("fluss-testing-hms-paimon");
+ warehousePath = baseDir.resolve("warehouse").toString();
+ customTablePath =
baseDir.resolve("custom_lake_table_path").toUri().toString();
+ conf.setString(
+ "datalake.paimon.javax.jdo.option.ConnectionURL",
+ "jdbc:derby:memory:" + baseDir.resolve("metastore_db") +
";create=true");
+ } catch (Exception e) {
+ throw new FlussRuntimeException("Failed to create hive catalog
test path", e);
+ }
+ conf.setString("datalake.paimon.warehouse", warehousePath);
+ conf.setString(
+ "datalake.paimon.javax.jdo.option.ConnectionDriverName",
+ "org.apache.derby.jdbc.EmbeddedDriver");
+ conf.setString("datalake.paimon.datanucleus.schema.autoCreateAll",
"true");
+ conf.setString("datalake.paimon.hive.metastore.schema.verification",
"false");
+
+ paimonCatalog =
+ CatalogFactory.createCatalog(
+
CatalogContext.create(Options.fromMap(extractLakeProperties(conf))));
Review Comment:
The test creates a static `paimonCatalog` but never closes it. Some catalog
implementations keep threads / open resources, which can lead to flaky builds
or hanging test JVMs. Add an `@AfterAll` (or equivalent) to close
`paimonCatalog` when the test class finishes to ensure deterministic cleanup.
##########
pom.xml:
##########
@@ -105,6 +105,12 @@
<paimon.version>1.3.1</paimon.version>
<iceberg.version>1.10.1</iceberg.version>
<hudi.version>1.1.0</hudi.version>
+ <hive.version>2.3.10</hive.version>
+ <datanucleus-api-jdo.version>4.2.4</datanucleus-api-jdo.version>
+ <datanucleus-core.version>4.1.17</datanucleus-core.version>
+ <datanucleus-rdbms.version>4.1.19</datanucleus-rdbms.version>
+ <javax-jdo.version>3.2.0-m3</javax-jdo.version>
+ <derby.version>10.14.2.0</derby.version>
Review Comment:
Using a milestone artifact version (`3.2.0-m3`) for `javax.jdo` can reduce
build reproducibility and may be less stable than a final release. Consider
switching to a stable release version (if available/compatible) to avoid
transient resolution issues and reduce dependency risk, especially since these
are brought in to support 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
