[flink-table-store] 05/06: [FLINK-25628] Introduce SstPathFactory

[lz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git

commit 9182e60c8ba51616de2905fd0dbb6e2d4b66d055
Author: JingsongLi <lz...@aliyun.com>
AuthorDate: Wed Jan 12 19:51:13 2022 +0800

    [FLINK-25628] Introduce SstPathFactory
    
    Co-authored-by: tsreaper <ts...@gmail.com>
---
 .../store/file/mergetree/sst/SstPathFactory.java   | 59 +++++++++++++++++++
 .../file/mergetree/sst/SstPathFactoryTest.java     | 66 ++++++++++++++++++++++
 2 files changed, 125 insertions(+)

diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstPathFactory.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstPathFactory.java
new file mode 100644
index 0000000..cfda5c3
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstPathFactory.java
@@ -0,0 +1,59 @@
+/*
+ * 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.flink.table.store.file.mergetree.sst;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.core.fs.Path;
+
+import javax.annotation.Nullable;
+
+import java.util.UUID;
+
+/** Factory which produces new {@link Path}s for sst files. */
+public class SstPathFactory {
+
+    private final Path bucketDir;
+    private final String uuid;
+
+    private int pathCount;
+
+    public SstPathFactory(Path root, @Nullable String partition, int bucket) {
+        if (partition == null) {
+            this.bucketDir = new Path(root + "/bucket-" + bucket);
+        } else {
+            this.bucketDir = new Path(root + "/" + partition + "/bucket-" + bucket);
+        }
+        this.uuid = UUID.randomUUID().toString();
+
+        this.pathCount = 0;
+    }
+
+    public Path newPath() {
+        return new Path(bucketDir + "/sst-" + uuid + "-" + (pathCount++));
+    }
+
+    public Path toPath(String fileName) {
+        return new Path(bucketDir + "/" + fileName);
+    }
+
+    @VisibleForTesting
+    public String uuid() {
+        return uuid;
+    }
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/sst/SstPathFactoryTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/sst/SstPathFactoryTest.java
new file mode 100644
index 0000000..315e85c
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/sst/SstPathFactoryTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.flink.table.store.file.mergetree.sst;
+
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link SstPathFactory}. */
+public class SstPathFactoryTest {
+
+    @TempDir java.nio.file.Path tempDir;
+
+    @Test
+    public void testNoPartition() {
+        SstPathFactory pathFactory = new SstPathFactory(new Path(tempDir.toString()), null, 123);
+        String uuid = pathFactory.uuid();
+
+        for (int i = 0; i < 20; i++) {
+            assertThat(pathFactory.newPath())
+                    .isEqualTo(new Path(tempDir.toString() + "/bucket-123/sst-" + uuid + "-" + i));
+        }
+        assertThat(pathFactory.toPath("my-sst-file-name"))
+                .isEqualTo(new Path(tempDir.toString() + "/bucket-123/my-sst-file-name"));
+    }
+
+    @Test
+    public void testWithPartition() {
+        SstPathFactory pathFactory =
+                new SstPathFactory(new Path(tempDir.toString()), "dt=20211224", 123);
+        String uuid = pathFactory.uuid();
+
+        for (int i = 0; i < 20; i++) {
+            assertThat(pathFactory.newPath())
+                    .isEqualTo(
+                            new Path(
+                                    tempDir.toString()
+                                            + "/dt=20211224/bucket-123/sst-"
+                                            + uuid
+                                            + "-"
+                                            + i));
+        }
+        assertThat(pathFactory.toPath("my-sst-file-name"))
+                .isEqualTo(
+                        new Path(tempDir.toString() + "/dt=20211224/bucket-123/my-sst-file-name"));
+    }
+}