Skip to content

[server] Update Tablet Server to use per-table/partition remote data directory #2782

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LiebingYu wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[server] Update Tablet Server to use per-table/partition remote data directory #2782

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 34 additions & 2 deletions fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.fluss.remote;

import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;

Expand All @@ -38,13 +40,17 @@ public class RemoteLogManifest {
private final TableBucket tableBucket;
private final List<RemoteLogSegment> remoteLogSegmentList;

private final FsPath remoteLogDir;

public RemoteLogManifest(
PhysicalTablePath physicalTablePath,
TableBucket tableBucket,
List<RemoteLogSegment> remoteLogSegmentList) {
List<RemoteLogSegment> remoteLogSegmentList,
FsPath remoteLogDir) {
this.physicalTablePath = physicalTablePath;
this.tableBucket = tableBucket;
this.remoteLogSegmentList = Collections.unmodifiableList(remoteLogSegmentList);
this.remoteLogDir = remoteLogDir;

// sanity check
for (RemoteLogSegment remoteLogSegment : remoteLogSegmentList) {
Expand Down Expand Up @@ -73,7 +79,7 @@ public RemoteLogManifest trimAndMerge(
}
newSegments.addAll(addedSegments);
newSegments.sort(Comparator.comparingLong(RemoteLogSegment::remoteLogStartOffset));
return new RemoteLogManifest(physicalTablePath, tableBucket, newSegments);
return new RemoteLogManifest(physicalTablePath, tableBucket, newSegments, remoteLogDir);
}

public long getRemoteLogStartOffset() {
Expand Down Expand Up @@ -120,10 +126,36 @@ public TableBucket getTableBucket() {
return tableBucket;
}

public FsPath getRemoteLogDir() {
return remoteLogDir;
}

@VisibleForTesting
public List<RemoteLogSegment> getRemoteLogSegmentList() {
return remoteLogSegmentList;
}

public RemoteLogManifest newManifest(FsPath remoteLogDir) {
List<RemoteLogSegment> newRemoteLogSegments = new ArrayList<>(remoteLogSegmentList.size());
for (RemoteLogSegment remoteLogSegment : remoteLogSegmentList) {
newRemoteLogSegments.add(
RemoteLogSegment.Builder.builder()
.physicalTablePath(remoteLogSegment.physicalTablePath())
.tableBucket(remoteLogSegment.tableBucket())
.remoteLogSegmentId(remoteLogSegment.remoteLogSegmentId())
.remoteLogStartOffset(remoteLogSegment.remoteLogStartOffset())
.remoteLogEndOffset(remoteLogSegment.remoteLogEndOffset())
.maxTimestamp(remoteLogSegment.maxTimestamp())
.segmentSizeInBytes(remoteLogSegment.segmentSizeInBytes())
// We set remoteLogDir manually here, so subsequent usage will be safe
// to directly use it.
.remoteLogDir(remoteLogDir)
.build());
}
return new RemoteLogManifest(
physicalTablePath, tableBucket, newRemoteLogSegments, remoteLogDir);
}

@Override
public boolean equals(Object o) {
if (this == o) {
Expand Down
21 changes: 20 additions & 1 deletion fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestJsonSerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.fluss.remote;

import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
Expand Down Expand Up @@ -49,6 +50,7 @@ public class RemoteLogManifestJsonSerde
private static final String END_OFFSET_FIELD = "end_offset";
private static final String MAX_TIMESTAMP_FIELD = "max_timestamp";
private static final String SEGMENT_SIZE_IN_BYTES_FIELD = "size_in_bytes";
private static final String REMOTE_LOG_DIR_FIELD = "remote_log_dir";
private static final int SNAPSHOT_VERSION = 1;

@Override
Expand Down Expand Up @@ -84,9 +86,14 @@ public void serialize(RemoteLogManifest manifest, JsonGenerator generator) throw
generator.writeNumberField(MAX_TIMESTAMP_FIELD, remoteLogSegment.maxTimestamp());
generator.writeNumberField(
SEGMENT_SIZE_IN_BYTES_FIELD, remoteLogSegment.segmentSizeInBytes());
generator.writeStringField(
REMOTE_LOG_DIR_FIELD, remoteLogSegment.remoteLogDir().toString());
generator.writeEndObject();
}
generator.writeEndArray();

generator.writeStringField(REMOTE_LOG_DIR_FIELD, manifest.getRemoteLogDir().toString());

generator.writeEndObject();
}

Expand Down Expand Up @@ -118,6 +125,11 @@ public RemoteLogManifest deserialize(JsonNode node) {
long endOffset = entryJson.get(END_OFFSET_FIELD).asLong();
long maxTimestamp = entryJson.get(MAX_TIMESTAMP_FIELD).asLong();
int segmentSizeInBytes = entryJson.get(SEGMENT_SIZE_IN_BYTES_FIELD).asInt();
// backward compatibility for existing RemoteLogSegment which does not have remoteLogDir
FsPath remoteLogDir = null;
if (entryJson.has(REMOTE_LOG_DIR_FIELD)) {
remoteLogDir = new FsPath(entryJson.get(REMOTE_LOG_DIR_FIELD).asText());
}
snapshotEntries.add(
RemoteLogSegment.Builder.builder()
.physicalTablePath(physicalTablePath)
Expand All @@ -127,10 +139,17 @@ public RemoteLogManifest deserialize(JsonNode node) {
.remoteLogEndOffset(endOffset)
.maxTimestamp(maxTimestamp)
.segmentSizeInBytes(segmentSizeInBytes)
.remoteLogDir(remoteLogDir)
.build());
}

return new RemoteLogManifest(physicalTablePath, tableBucket, snapshotEntries);
// backward compatibility for existing RemoteLogManifest which does not have remoteLogDir
FsPath remoteLogDir = null;
if (node.has(REMOTE_LOG_DIR_FIELD)) {
remoteLogDir = new FsPath(node.get(REMOTE_LOG_DIR_FIELD).asText());
}

return new RemoteLogManifest(physicalTablePath, tableBucket, snapshotEntries, remoteLogDir);
}

public static RemoteLogManifest fromJson(byte[] json) {
Expand Down
28 changes: 24 additions & 4 deletions fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.fluss.remote;

import org.apache.fluss.annotation.Internal;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;

Expand Down Expand Up @@ -50,14 +51,17 @@ public class RemoteLogSegment {

private final int segmentSizeInBytes;

private final FsPath remoteLogDir;

private RemoteLogSegment(
PhysicalTablePath physicalTablePath,
TableBucket tableBucket,
UUID remoteLogSegmentId,
long remoteLogStartOffset,
long remoteLogEndOffset,
long maxTimestamp,
int segmentSizeInBytes) {
int segmentSizeInBytes,
FsPath remoteLogDir) {
Comment thread
LiebingYu marked this conversation as resolved.
this.physicalTablePath = checkNotNull(physicalTablePath);
this.tableBucket = checkNotNull(tableBucket);
this.remoteLogSegmentId = checkNotNull(remoteLogSegmentId);
Expand All @@ -79,6 +83,7 @@ private RemoteLogSegment(
this.remoteLogEndOffset = remoteLogEndOffset;
this.maxTimestamp = maxTimestamp;
this.segmentSizeInBytes = segmentSizeInBytes;
this.remoteLogDir = remoteLogDir;
}

public PhysicalTablePath physicalTablePath() {
Expand Down Expand Up @@ -115,6 +120,10 @@ public int segmentSizeInBytes() {
return segmentSizeInBytes;
}

public FsPath remoteLogDir() {
return remoteLogDir;
}

@Override
public boolean equals(Object o) {
if (this == o) {
Expand All @@ -130,7 +139,8 @@ public boolean equals(Object o) {
&& maxTimestamp == that.maxTimestamp
&& Objects.equals(remoteLogSegmentId, that.remoteLogSegmentId)
&& Objects.equals(physicalTablePath, that.physicalTablePath)
&& Objects.equals(tableBucket, that.tableBucket);
&& Objects.equals(tableBucket, that.tableBucket)
&& Objects.equals(remoteLogDir, that.remoteLogDir);
}

@Override
Expand All @@ -142,7 +152,8 @@ public int hashCode() {
remoteLogStartOffset,
remoteLogEndOffset,
maxTimestamp,
segmentSizeInBytes);
segmentSizeInBytes,
remoteLogDir);
}

@Override
Expand All @@ -162,6 +173,8 @@ public String toString() {
+ maxTimestamp
+ ", segmentSizeInBytes="
+ segmentSizeInBytes
+ ", remoteLogDir="
+ remoteLogDir
+ '}';
}

Expand All @@ -174,6 +187,7 @@ public static class Builder {
private long remoteLogEndOffset;
private long maxTimestamp;
private int segmentSizeInBytes;
private FsPath remoteLogDir;

public static Builder builder() {
return new Builder();
Expand Down Expand Up @@ -214,6 +228,11 @@ public Builder tableBucket(TableBucket tableBucket) {
return this;
}

public Builder remoteLogDir(FsPath remoteLogDir) {
this.remoteLogDir = remoteLogDir;
return this;
}

public RemoteLogSegment build() {
return new RemoteLogSegment(
physicalTablePath,
Expand All @@ -222,7 +241,8 @@ public RemoteLogSegment build() {
remoteLogStartOffset,
remoteLogEndOffset,
maxTimestamp,
segmentSizeInBytes);
segmentSizeInBytes,
remoteLogDir);
}
}
}
26 changes: 26 additions & 0 deletions fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,6 +426,19 @@ public static FsPath remoteLogDir(Configuration conf) {
return new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR) + "/" + REMOTE_LOG_DIR_NAME);
}

/**
* Returns the remote root directory path for storing log files.
*
* <p>The path contract:
*
* <pre>
* {$remote.data.dir}/log
* </pre>
*/
public static FsPath remoteLogDir(String remoteDataDir) {
return new FsPath(remoteDataDir, REMOTE_LOG_DIR_NAME);
}

/**
* Returns the remote directory path for storing log files for a log tablet.
*
Expand Down Expand Up @@ -592,6 +605,19 @@ public static FsPath remoteKvDir(Configuration conf) {
return new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR) + "/" + REMOTE_KV_DIR_NAME);
}

/**
* Returns the remote root directory path for storing kv snapshot files.
*
* <p>The path contract:
*
* <pre>
* {$remote.data.dir}/kv
* </pre>
*/
public static FsPath remoteKvDir(String remoteDataDir) {
return new FsPath(remoteDataDir, REMOTE_KV_DIR_NAME);
}

/**
* Returns the remote directory path for storing kv snapshot files for a kv tablet.
*
Expand Down
Loading
Loading