[connector] Support spark catalog and introduce some basic classes to support spark read and write

Alibaba-HZY · Alibaba-HZY · commit ea205d2d868b · 2025-09-03T17:42:43.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java b/fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java
@@ -147,9 +147,7 @@ public ServerNode getCoordinatorServer() {
         return coordinatorServer;
     }
 
-    /**
-     * @return The known set of alive tablet servers.
-     */
+    /** @return The known set of alive tablet servers. */
     public Map<Integer, ServerNode> getAliveTabletServers() {
         return aliveTabletServersById;
     }
diff --git a/fluss-common/src/main/java/org/apache/fluss/memory/MemorySegmentPool.java b/fluss-common/src/main/java/org/apache/fluss/memory/MemorySegmentPool.java
@@ -66,14 +66,10 @@ public interface MemorySegmentPool {
      */
     void returnAll(List<MemorySegment> memory);
 
-    /**
-     * @return Free page number.
-     */
+    /** @return Free page number. */
     int freePages();
 
-    /**
-     * @return the available memory size in bytes.
-     */
+    /** @return the available memory size in bytes. */
     long availableMemory();
 
     void close();
diff --git a/fluss-common/src/main/java/org/apache/fluss/predicate/Predicate.java b/fluss-common/src/main/java/org/apache/fluss/predicate/Predicate.java
@@ -47,9 +47,7 @@ public interface Predicate extends Serializable {
      */
     boolean test(long rowCount, InternalRow minValues, InternalRow maxValues, Long[] nullCounts);
 
-    /**
-     * @return the negation predicate of this predicate if possible.
-     */
+    /** @return the negation predicate of this predicate if possible. */
     Optional<Predicate> negate();
 
     <T> T visit(PredicateVisitor<T> visitor);
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java b/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java
@@ -257,9 +257,7 @@ public void append(short schemaId, BinaryRow row) throws IOException {
             currentRecordNumber++;
         }
 
-        /**
-         * @param valueBytes consisted of schema id and the row encoded in the value bytes
-         */
+        /** @param valueBytes consisted of schema id and the row encoded in the value bytes */
         public void append(byte[] valueBytes) throws IOException {
             if (isClosed) {
                 throw new IllegalStateException(
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/send/SendWritableOutput.java b/fluss-common/src/main/java/org/apache/fluss/record/send/SendWritableOutput.java
@@ -34,9 +34,7 @@ public class SendWritableOutput extends ByteBufWritableOutput {
     /** The current reader index of the underlying {@link #buf} for building next {@link Send}. */
     private int currentReaderIndex = 0;
 
-    /**
-     * @param buf The ByteBuf that has capacity of data size excluding zero-copy.
-     */
+    /** @param buf The ByteBuf that has capacity of data size excluding zero-copy. */
     public SendWritableOutput(ByteBuf buf) {
         super(buf);
         this.sends = new ArrayDeque<>(1);
diff --git a/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java b/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java
@@ -93,16 +93,12 @@ public UUID remoteLogSegmentId() {
         return remoteLogSegmentId;
     }
 
-    /**
-     * @return Remote log start offset of this segment (inclusive).
-     */
+    /** @return Remote log start offset of this segment (inclusive). */
     public long remoteLogStartOffset() {
         return remoteLogStartOffset;
     }
 
-    /**
-     * @return Remote log end offset of this segment (inclusive).
-     */
+    /** @return Remote log end offset of this segment (inclusive). */
     public long remoteLogEndOffset() {
         return remoteLogEndOffset;
     }
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/types/Either.java b/fluss-common/src/main/java/org/apache/fluss/utils/types/Either.java
@@ -60,16 +60,12 @@ public static <L, R> Either<L, R> right(R value) {
      */
     public abstract R right() throws IllegalStateException;
 
-    /**
-     * @return true if this is a Left value, false if this is a Right value
-     */
+    /** @return true if this is a Left value, false if this is a Right value */
     public final boolean isLeft() {
         return getClass() == Left.class;
     }
 
-    /**
-     * @return true if this is a Right value, false if this is a Left value
-     */
+    /** @return true if this is a Right value, false if this is a Left value */
     public final boolean isRight() {
         return getClass() == Right.class;
     }
diff --git a/fluss-common/src/test/java/org/apache/fluss/security/auth/AuthenticationFactoryTest.java b/fluss-common/src/test/java/org/apache/fluss/security/auth/AuthenticationFactoryTest.java
@@ -80,8 +80,7 @@ void testIdentifierCaseInsensitive() {
                 .isInstanceOf(TestIdentifierClientAuthenticator.class);
         assertThat(
                         AuthenticationFactory.loadServerAuthenticatorSuppliers(configuration)
-                                .values()
-                                .stream()
+                                .values().stream()
                                 .findAny()
                                 .get()
                                 .get())
@@ -94,8 +93,7 @@ void testIdentifierCaseInsensitive() {
                 .isInstanceOf(TestIdentifierClientAuthenticator.class);
         assertThat(
                         AuthenticationFactory.loadServerAuthenticatorSuppliers(configuration)
-                                .values()
-                                .stream()
+                                .values().stream()
                                 .findAny()
                                 .get()
                                 .get())
diff --git a/fluss-filesystems/fluss-fs-gs/src/test/java/org/apache/fluss/fs/gs/GSFileSystemBehaviorITCase.java b/fluss-filesystems/fluss-fs-gs/src/test/java/org/apache/fluss/fs/gs/GSFileSystemBehaviorITCase.java
@@ -93,7 +93,8 @@ private static void applyInMemoryStorage(FileSystem fileSystem) throws IOExcepti
             GoogleCloudStorageFileSystem googleCloudStorageFileSystem =
                     new GoogleCloudStorageFileSystem(
                             googleCloudStorageOptions -> inMemoryGoogleCloudStorage,
-                            GoogleCloudStorageFileSystemOptions.DEFAULT.toBuilder()
+                            GoogleCloudStorageFileSystemOptions.DEFAULT
+                                    .toBuilder()
                                     .setCloudStorageOptions(inMemoryGoogleCloudStorage.getOptions())
                                     .build());
 
diff --git a/fluss-filesystems/fluss-fs-s3/src/main/java/com/amazonaws/services/s3/model/transform/XmlResponsesSaxParser.java b/fluss-filesystems/fluss-fs-s3/src/main/java/com/amazonaws/services/s3/model/transform/XmlResponsesSaxParser.java
@@ -924,16 +924,12 @@ public static class ListAllMyBucketsHandler extends AbstractHandler {
 
         private Bucket currentBucket = null;
 
-        /**
-         * @return the buckets listed in the document.
-         */
+        /** @return the buckets listed in the document. */
         public List<Bucket> getBuckets() {
             return buckets;
         }
 
-        /**
-         * @return the owner of the buckets.
-         */
+        /** @return the owner of the buckets. */
         public Owner getOwner() {
             return bucketsOwner;
         }
@@ -990,9 +986,7 @@ public static class AccessControlListHandler extends AbstractHandler {
         private Grantee currentGrantee = null;
         private Permission currentPermission = null;
 
-        /**
-         * @return an object representing the ACL document.
-         */
+        /** @return an object representing the ACL document. */
         public AccessControlList getAccessControlList() {
             return accessControlList;
         }
@@ -1073,9 +1067,7 @@ public static class BucketLoggingConfigurationHandler extends AbstractHandler {
         private final BucketLoggingConfiguration bucketLoggingConfiguration =
                 new BucketLoggingConfiguration();
 
-        /**
-         * @return an object representing the bucket's LoggingStatus document.
-         */
+        /** @return an object representing the bucket's LoggingStatus document. */
         public BucketLoggingConfiguration getBucketLoggingConfiguration() {
             return bucketLoggingConfiguration;
         }
@@ -1105,9 +1097,7 @@ public static class BucketLocationHandler extends AbstractHandler {
 
         private String location = null;
 
-        /**
-         * @return the bucket's location.
-         */
+        /** @return the bucket's location. */
         public String getLocation() {
             return location;
         }
diff --git a/fluss-flink/fluss-flink-2.1/src/main/java/org/apache/flink/api/connector/sink2/Sink.java b/fluss-flink/fluss-flink-2.1/src/main/java/org/apache/flink/api/connector/sink2/Sink.java
@@ -53,9 +53,7 @@ interface InitContext {
          */
         MailboxExecutor getMailboxExecutor();
 
-        /**
-         * @return The metric group this writer belongs to.
-         */
+        /** @return The metric group this writer belongs to. */
         SinkWriterMetricGroup metricGroup();
     }
 }
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergConversionsTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergConversionsTest.java
@@ -44,8 +44,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** UT for {@link IcebergConversions}. */
-class IcebergConversionsTest {
-    ;
+class IcebergConversionsTest {;
 
     @Test
     void testToPartition(@TempDir File tempWarehouseDir) {
diff --git a/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/BytesTest.java b/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/BytesTest.java
@@ -62,7 +62,8 @@ public void testBytes() throws Exception {
 
         // test binary equals to protobuf
         Bytes.B pbb =
-                Bytes.B.newBuilder()
+                Bytes.B
+                        .newBuilder()
                         .setPayload(ByteString.copyFrom(new byte[] {1, 2, 3, 4, 5}))
                         .build();
 
@@ -254,7 +255,8 @@ public void testRepeatedBytes() throws Exception {
         assertThat(lpb.getExtraItemAt(1)).isEqualTo(new byte[] {4, 5, 6, 7});
 
         Bytes.B pbb =
-                Bytes.B.newBuilder()
+                Bytes.B
+                        .newBuilder()
                         .addExtraItems(ByteString.copyFrom(new byte[] {1, 2, 3}))
                         .addExtraItems(ByteString.copyFrom(new byte[] {4, 5, 6, 7}))
                         .build();
diff --git a/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/StringsTest.java b/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/StringsTest.java
@@ -58,7 +58,8 @@ public void testStrings() throws Exception {
         assertThat(lps.getNameAt(2)).isEqualTo("c");
 
         Strings.S pbs =
-                Strings.S.newBuilder()
+                Strings.S
+                        .newBuilder()
                         .setId("id")
                         .addNames("a")
                         .addNames("b")
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/NettyUtils.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/NettyUtils.java
@@ -36,9 +36,7 @@
 /** Utils of netty. */
 public class NettyUtils {
 
-    /**
-     * @return an EventLoopGroup suitable for the current platform
-     */
+    /** @return an EventLoopGroup suitable for the current platform */
     public static EventLoopGroup newEventLoopGroup(int nThreads, String threadNamePrefix) {
         if (Epoll.isAvailable()) {
             // Regular Epoll based event loop
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogSegments.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogSegments.java
@@ -175,30 +175,22 @@ public Optional<LogSegment> floorSegment(long offset) {
         return floorEntry(offset).map(Map.Entry::getValue);
     }
 
-    /**
-     * @return the entry associated with the greatest offset, if it exists.
-     */
+    /** @return the entry associated with the greatest offset, if it exists. */
     public Optional<Map.Entry<Long, LogSegment>> lastEntry() {
         return Optional.ofNullable(segments.lastEntry());
     }
 
-    /**
-     * @return the log segment with the greatest offset, if it exists.
-     */
+    /** @return the log segment with the greatest offset, if it exists. */
     public Optional<LogSegment> lastSegment() {
         return Optional.ofNullable(segments.lastEntry()).map(Map.Entry::getValue);
     }
 
-    /**
-     * @return the entry associated with the greatest offset, if it exists.
-     */
+    /** @return the entry associated with the greatest offset, if it exists. */
     public Optional<Map.Entry<Long, LogSegment>> firstEntry() {
         return Optional.ofNullable(segments.firstEntry());
     }
 
-    /**
-     * @return the log segment with the greatest offset, if it exists.
-     */
+    /** @return the log segment with the greatest offset, if it exists. */
     public Optional<LogSegment> firstSegment() {
         return Optional.ofNullable(segments.firstEntry()).map(Map.Entry::getValue);
     }
diff --git a/fluss-test-utils/src/main/java/org/apache/fluss/testutils/common/FlussAssertions.java b/fluss-test-utils/src/main/java/org/apache/fluss/testutils/common/FlussAssertions.java
@@ -35,9 +35,7 @@ public class FlussAssertions {
 
     private FlussAssertions() {}
 
-    /**
-     * @see #chainOfCauses(Throwable)
-     */
+    /** @see #chainOfCauses(Throwable) */
     @SuppressWarnings({"rawtypes", "unused"})
     public static final InstanceOfAssertFactory<Stream, ListAssert<Throwable>> STREAM_THROWABLE =
             new InstanceOfAssertFactory<>(Stream.class, Assertions::<Throwable>assertThat);
diff --git a/tools/ci/fluss-ci-tools/src/main/java/org/apache/fluss/tools/ci/licensecheck/NoticeFileChecker.java b/tools/ci/fluss-ci-tools/src/main/java/org/apache/fluss/tools/ci/licensecheck/NoticeFileChecker.java
@@ -304,8 +304,7 @@ static Map<Severity, List<String>> checkNoticeFile(
 
             // find all dependencies missing from NOTICE file
             Collection<Dependency> expectedDependencies =
-                    modulesWithShadedDependencies
-                            .getOrDefault(moduleName, Collections.emptySet())
+                    modulesWithShadedDependencies.getOrDefault(moduleName, Collections.emptySet())
                             .stream()
                             .filter(
                                     dependency ->
diff --git a/tools/ci/fluss-ci-tools/src/main/java/org/apache/fluss/tools/ci/utils/deploy/DeployParser.java b/tools/ci/fluss-ci-tools/src/main/java/org/apache/fluss/tools/ci/utils/deploy/DeployParser.java
@@ -58,8 +58,7 @@ public static Set<String> parseDeployOutput(File buildResult) throws IOException
     static Set<String> parseDeployOutput(Stream<String> lines) {
         return ParserUtils.parsePluginOutput(
                         lines, DEPLOY_MODULE_PATTERN, DeployParser::parseDeployBlock)
-                .entrySet()
-                .stream()
+                .entrySet().stream()
                 .filter(Map.Entry::getValue)
                 .map(Map.Entry::getKey)
                 .collect(Collectors.toSet());

Original file line number	Diff line number	Diff line change
`@@ -147,9 +147,7 @@ public ServerNode getCoordinatorServer() {`
`147`	`147`	`return coordinatorServer;`
`148`	`148`	`}`
`149`	`149`
`150`		`- /**`
`151`		`- * @return The known set of alive tablet servers.`
`152`		`- */`
	`150`	`+ /** @return The known set of alive tablet servers. */`
`153`	`151`	`public Map<Integer, ServerNode> getAliveTabletServers() {`
`154`	`152`	`return aliveTabletServersById;`
`155`	`153`	`}`
Original file line number	Diff line number	Diff line change
`@@ -93,16 +93,12 @@ public UUID remoteLogSegmentId() {`
`93`	`93`	`return remoteLogSegmentId;`
`94`	`94`	`}`
`95`	`95`
`96`		`- /**`
`97`		`- * @return Remote log start offset of this segment (inclusive).`
`98`		`- */`
	`96`	`+ /** @return Remote log start offset of this segment (inclusive). */`
`99`	`97`	`public long remoteLogStartOffset() {`
`100`	`98`	`return remoteLogStartOffset;`
`101`	`99`	`}`
`102`	`100`
`103`		`- /**`
`104`		`- * @return Remote log end offset of this segment (inclusive).`
`105`		`- */`
	`101`	`+ /** @return Remote log end offset of this segment (inclusive). */`
`106`	`102`	`public long remoteLogEndOffset() {`
`107`	`103`	`return remoteLogEndOffset;`
`108`	`104`	`}`