wuchong commented on code in PR #2037: URL: https://github.com/apache/fluss/pull/2037#discussion_r2629981139
########## fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.server.zk.data.lake; + +import org.apache.fluss.fs.FSDataInputStream; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.server.zk.data.ZkData; +import org.apache.fluss.utils.IOUtils; + +import javax.annotation.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Collections; +import java.util.List; + +import static org.apache.fluss.metrics.registry.MetricRegistry.LOG; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * Represents lake table snapshot information stored in {@link ZkData.LakeTableZNode}. + * + * <p>This class supports two storage formats: + * + * <ul> + * <li>Version 1 (legacy): Contains the full {@link LakeTableSnapshot} data directly + * <li>Version 2 (current): Contains a list of lake snapshot, recording the metadata file path for + * different lake snapshots, with actual metadata storing in file to reduce zk pressure + * </ul> + * + * @see LakeTableJsonSerde for JSON serialization and deserialization + */ +public class LakeTable { + + // Version 2 (current): + // a list of lake snapshot metadata, record the metadata for different lake snapshots + @Nullable private final List<LakeSnapshotMetadata> lakeSnapshotMetadata; + + // Version 1 (legacy): the full lake table snapshot info stored in ZK, will be null in version2 + @Nullable private final LakeTableSnapshot lakeTableSnapshot; + + /** + * Creates a LakeTable from a LakeTableSnapshot (version 1 format). + * + * @param lakeTableSnapshot the snapshot data + */ + public LakeTable(LakeTableSnapshot lakeTableSnapshot) { + this(lakeTableSnapshot, null); + } + + /** + * Creates a LakeTable with a lake snapshot metadata (version 2 format). + * + * @param lakeSnapshotMetadata the metadata containing the file path to the snapshot data + */ + public LakeTable(LakeSnapshotMetadata lakeSnapshotMetadata) { + this(null, Collections.singletonList(lakeSnapshotMetadata)); + } + + /** + * Creates a LakeTable with a list of lake snapshot metadata (version 2 format). + * + * @param lakeSnapshotMetadata the list of lake snapshot metadata + */ + public LakeTable(List<LakeSnapshotMetadata> lakeSnapshotMetadata) { + this(null, lakeSnapshotMetadata); + } + + private LakeTable( + @Nullable LakeTableSnapshot lakeTableSnapshot, + List<LakeSnapshotMetadata> lakeSnapshotMetadata) { + this.lakeTableSnapshot = lakeTableSnapshot; + this.lakeSnapshotMetadata = lakeSnapshotMetadata; + } + + @Nullable + public LakeSnapshotMetadata getLakeTableLatestSnapshot() { Review Comment: The method name confuses a lot, because it doesn't returns a `LakeTableSnapshot`. Better to name `getLatestLakeSnapshotMetadata`. ########## fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerde.java: ########## @@ -0,0 +1,123 @@ +/* + * 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.server.zk.data.lake; + +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.fluss.utils.json.JsonDeserializer; +import org.apache.fluss.utils.json.JsonSerializer; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * Json serializer and deserializer for {@link LakeTable}. + * + * <p>This serde supports two storage format versions: + * + * <ul> + * <li>Version 1 (legacy): ZK node contains full {@link LakeTableSnapshot} data. During + * deserialization, it uses {@link LakeTableSnapshotJsonSerde} to deserialize and wraps the + * result in a {@link LakeTable}. + * <li>Version 2 (current): ZK node contains only the lake table snapshot file paths. The actual + * snapshot data is stored in a remote file pointed by the lake table snapshot file path. + * </ul> + */ +public class LakeTableJsonSerde implements JsonSerializer<LakeTable>, JsonDeserializer<LakeTable> { Review Comment: I didn't see a `LakeTableJsonSerdeTest` for this class. We should have such test to guaratneee the JSON format is as expected, and also for future compatibility tests. ########## fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.server.zk.data.lake; + +import org.apache.fluss.fs.FSDataInputStream; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.server.zk.data.ZkData; +import org.apache.fluss.utils.IOUtils; + +import javax.annotation.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Collections; +import java.util.List; + +import static org.apache.fluss.metrics.registry.MetricRegistry.LOG; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * Represents lake table snapshot information stored in {@link ZkData.LakeTableZNode}. + * + * <p>This class supports two storage formats: + * + * <ul> + * <li>Version 1 (legacy): Contains the full {@link LakeTableSnapshot} data directly + * <li>Version 2 (current): Contains a list of lake snapshot, recording the metadata file path for + * different lake snapshots, with actual metadata storing in file to reduce zk pressure + * </ul> + * + * @see LakeTableJsonSerde for JSON serialization and deserialization + */ +public class LakeTable { + + // Version 2 (current): + // a list of lake snapshot metadata, record the metadata for different lake snapshots + @Nullable private final List<LakeSnapshotMetadata> lakeSnapshotMetadata; + + // Version 1 (legacy): the full lake table snapshot info stored in ZK, will be null in version2 + @Nullable private final LakeTableSnapshot lakeTableSnapshot; + + /** + * Creates a LakeTable from a LakeTableSnapshot (version 1 format). + * + * @param lakeTableSnapshot the snapshot data + */ + public LakeTable(LakeTableSnapshot lakeTableSnapshot) { + this(lakeTableSnapshot, null); + } + + /** + * Creates a LakeTable with a lake snapshot metadata (version 2 format). + * + * @param lakeSnapshotMetadata the metadata containing the file path to the snapshot data + */ + public LakeTable(LakeSnapshotMetadata lakeSnapshotMetadata) { + this(null, Collections.singletonList(lakeSnapshotMetadata)); + } + + /** + * Creates a LakeTable with a list of lake snapshot metadata (version 2 format). + * + * @param lakeSnapshotMetadata the list of lake snapshot metadata + */ + public LakeTable(List<LakeSnapshotMetadata> lakeSnapshotMetadata) { + this(null, lakeSnapshotMetadata); + } + + private LakeTable( + @Nullable LakeTableSnapshot lakeTableSnapshot, + List<LakeSnapshotMetadata> lakeSnapshotMetadata) { + this.lakeTableSnapshot = lakeTableSnapshot; + this.lakeSnapshotMetadata = lakeSnapshotMetadata; + } + + @Nullable + public LakeSnapshotMetadata getLakeTableLatestSnapshot() { + if (lakeSnapshotMetadata != null && !lakeSnapshotMetadata.isEmpty()) { + return lakeSnapshotMetadata.get(0); + } + return null; + } + + @Nullable + public List<LakeSnapshotMetadata> getLakeSnapshotMetadata() { Review Comment: It returns a list of `LakeSnapshotMetadata`, better to name `getLakeSnapshotMetadatas`. ########## fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java: ########## @@ -769,13 +786,19 @@ public class ConfigOptions { "Size of the thread pool used in scheduling tasks to copy segments, " + "fetch remote log indexes and clean up remote log segments."); + /** + * @deprecated This option is deprecated. Please use {@link ConfigOptions#SERVER_IO_POOL_SIZE} + * instead. + */ + @Deprecated public static final ConfigOption<Integer> REMOTE_LOG_DATA_TRANSFER_THREAD_NUM = Review Comment: There is still a usage in the test code. We should update the test code to use the new config. -- 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]
