Merge branch 'apache:main' into issue-489-parition-key-more-types-apache

Author: MaggieSullivan

.github/workflows/docs-check.yaml

@@ -18,11 +18,17 @@
 
 # This workflow is meant for checking broken links in the documentation.
 name: Check Documentation
+permissions:
+  contents: read
 on:
   pull_request:
     branches: [main, release-*, ci-*]
     paths:
       - 'website/**'
+  push:
+    branches: [main, release-*, ci-*]
+    paths:
+      - 'website/**'
 
 jobs:
   test-deploy:

.github/workflows/docs-deploy.yaml

@@ -16,39 +16,27 @@
 # limitations under the License.
 ################################################################################
 name: Deploy Documentation
+permissions:
+  contents: read
 on:
   push:
     branches: [main, release-*]
     paths:
       - 'website/**'
 
+# We use `repository_dispatch` to trigger the deployment job on the apache/fluss-website repository.
+# https://github.com/apache/fluss-website/blob/main/.github/workflows/website-deploy.yaml
+# https://docs.github.com/en/actions/reference/events-that-trigger-workflows#repository_dispatch
 jobs:
   deploy:
     runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./website
     steps:
-      - uses: actions/checkout@v4
-        with:
-          ref: main
-          fetch-depth: 0
-      - name: Generate versioned docs
-        run: ./build_versioned_docs.sh
-      - uses: actions/setup-node@v4
-        with:
-          node-version: 18
-      - name: Install dependencies
-        run: npm install
-      - name: Build website
-        run: npm run build
-      - uses: webfactory/ssh-agent@v0.5.0
-        with:
-          ssh-private-key: ${{ secrets.GH_PAGES_DEPLOY }}
-      - name: Deploy website
-        env:
-          USE_SSH: true
+      - name: Send Event to Trigger Deploy
         run: |
-          git config --global user.email "actions@github.com"
-          git config --global user.name "gh-actions"
-          npm run deploy -- --skip-build
+          curl -L \
+            -X POST \
+            -H "Accept: application/vnd.github+json" \
+            -H "Authorization: Bearer ${{ secrets.GH_TOKEN }}" \
+            -H "X-GitHub-Api-Version: 2022-11-28" \
+            https://api.github.com/repos/apache/fluss-website/dispatches \
+            -d '{"event_type":"website-deploy","client_payload":{"repository": "apache/fluss"}}'

.github/workflows/license-check.yml

@@ -14,6 +14,8 @@
 # limitations under the License.
 
 name: Check License
+permissions:
+  contents: read
 
 on: [push, pull_request]
 
@@ -37,6 +39,7 @@ jobs:
         with:
           java-version: 8
           distribution: 'temurin'
+
       - name: Build
         run: |
           set -o pipefail

.github/workflows/stage.sh

@@ -27,7 +27,9 @@ fluss-flink/fluss-flink-1.20,\
 fluss-flink/fluss-flink-1.19,\
 fluss-flink/fluss-flink-1.18,\
 fluss-lake,\
-fluss-lake/fluss-lake-paimon
+fluss-lake/fluss-lake-paimon,\
+fluss-lake/fluss-lake-iceberg,\
+fluss-lake/fluss-lake-lance
 "
 
 function get_test_modules_for_stage() {

.idea/vcs.xml

@@ -3,25 +3,26 @@
       <source media="(prefers-color-scheme: dark)" srcset="website/static/img/logo/svg/white_color_logo.svg">
       <source media="(prefers-color-scheme: light)" srcset="website/static/img/logo/svg/colored_logo.svg">
       <!-- Fall back to version that works for dark and light mode -->
-      <img alt="Fluss logo" src="website/static/img/logo/svg/white_filled.svg">
+      <img alt="Apache Fluss logo" src="website/static/img/logo/svg/white_filled.svg">
     </picture>
 </p>
 
 <p align="center">
-  <a href="https://alibaba.github.io/fluss-docs/docs/intro/">Documentation</a> | <a href="https://alibaba.github.io/fluss-docs/docs/quickstart/flink/">QuickStart</a> | <a href="https://alibaba.github.io/fluss-docs/community/dev/ide-setup/">Development</a>
+  <a href="https://fluss.apache.org/docs/">Documentation</a> | <a href="https://fluss.apache.org/docs/quickstart/flink/">QuickStart</a> | <a href="https://fluss.apache.org/community/dev/ide-setup/">Development</a>
 </p>
 
 <p align="center">
-  <a href="https://github.com/alibaba/fluss/actions/workflows/ci.yaml"><img src="https://github.com/alibaba/fluss/actions/workflows/ci.yaml/badge.svg?branch=main" alt="CI"></a>
-  <a href="https://github.com/alibaba/fluss/blob/main/LICENSE"><img src="https://img.shields.io/badge/license-Apache%202-4EB1BA.svg" alt="License"></a>
-  <a href="https://join.slack.com/t/fluss-hq/shared_invite/zt-33wlna581-QAooAiCmnYboJS8D_JUcYw"><img src="https://img.shields.io/badge/slack-join_chat-brightgreen.svg?logo=slack" alt="Slack"></a>
+  <a href="https://github.com/apache/fluss/actions/workflows/ci.yaml"><img src="https://github.com/apache/fluss/actions/workflows/ci.yaml/badge.svg?branch=main" alt="CI"></a>
+  <a href="https://github.com/apache/fluss/blob/main/LICENSE"><img src="https://img.shields.io/badge/license-Apache%202-4EB1BA.svg" alt="License"></a>
+  <a href="https://join.slack.com/t/apache-fluss/shared_invite/zt-33wlna581-QAooAiCmnYboJS8D_JUcYw"><img src="https://img.shields.io/badge/slack-join_chat-brightgreen.svg?logo=slack" alt="Slack"></a>
 </p>
 
-## What is Fluss?
+## What is Apache Fluss (Incubating)?
 
-Fluss is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures.
+Apache Fluss (Incubating) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures.
 
-It bridges the gap between **data streaming** and **data Lakehouse** by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines like **Apache Flink**, while Apache Spark, and StarRocks are coming soon.
+It bridges the gap between **data streaming** and **data Lakehouse** by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines like **Apache Flink**, while 
+Apache Spark, and StarRocks are coming soon.
 
 **Fluss (German: river, pronounced `/flus/`)** enables streaming data continuously converging, distributing and flowing into lakes, like a river 🌊
 
@@ -36,27 +37,27 @@ It bridges the gap between **data streaming** and **data Lakehouse** by enabling
 
 ## Building
 
-Prerequisites for building Fluss:
+Prerequisites for building Apache Fluss:
 
 - Unix-like environment (we use Linux, Mac OS X, Cygwin, WSL)
 - Git
 - Maven (we require version >= 3.8.6)
 - Java 8 or 11
 
 ```bash
-git clone https://github.com/alibaba/fluss.git
+git clone https://github.com/apache/fluss.git
 cd fluss
 ./mvnw clean package -DskipTests
 ```
 
-Fluss is now installed in `build-target`. The build command uses Maven Wrapper (`mvnw`) which ensures the correct Maven version is used.
+Apache Fluss is now installed in `build-target`. The build command uses Maven Wrapper (`mvnw`) which ensures the correct Maven version is used.
 
 ## Contributing
 
-Fluss is open-source, and we’d love your help to keep it growing! Join the [discussions](https://github.com/alibaba/fluss/discussions),
-open [issues](https://github.com/alibaba/fluss/issues) if you find a bug or request features, contribute code and documentation,
- or help us improve the project in any way. All contributions are welcome!
+Apache Fluss (Incubating) is open-source, and we’d love your help to keep it growing! Join the [discussions](https://github.com/apache/fluss/discussions),
+open [issues](https://github.com/apache/fluss/issues) if you find a bug or request features, contribute code and documentation,
+or help us improve the project in any way. All contributions are welcome!
 
 ## License
 
-Fluss project is licensed under the [Apache License 2.0](https://github.com/alibaba/fluss/blob/main/LICENSE).
+Apache Fluss (Incubating) project is licensed under the [Apache License 2.0](https://github.com/apache/fluss/blob/main/LICENSE).

README.md

@@ -38,6 +38,7 @@
 import com.alibaba.fluss.rpc.GatewayClientProxy;
 import com.alibaba.fluss.rpc.RpcClient;
 import com.alibaba.fluss.rpc.gateway.AdminGateway;
+import com.alibaba.fluss.rpc.gateway.AdminReadOnlyGateway;
 import com.alibaba.fluss.rpc.gateway.TabletServerGateway;
 import com.alibaba.fluss.rpc.messages.CreateAclsRequest;
 import com.alibaba.fluss.rpc.messages.CreateDatabaseRequest;
@@ -99,13 +100,17 @@
 public class FlussAdmin implements Admin {
 
     private final AdminGateway gateway;
+    private final AdminReadOnlyGateway readOnlyGateway;
     private final MetadataUpdater metadataUpdater;
     private final RpcClient client;
 
     public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) {
         this.gateway =
                 GatewayClientProxy.createGatewayProxy(
                         metadataUpdater::getCoordinatorServer, client, AdminGateway.class);
+        this.readOnlyGateway =
+                GatewayClientProxy.createGatewayProxy(
+                        metadataUpdater::getRandomTabletServer, client, AdminGateway.class);
         this.metadataUpdater = metadataUpdater;
         this.client = client;
     }
@@ -119,7 +124,7 @@ public CompletableFuture<List<ServerNode>> getServerNodes() {
                         List<ServerNode> serverNodeList = new ArrayList<>();
                         Cluster cluster =
                                 sendMetadataRequestAndRebuildCluster(
-                                        gateway,
+                                        readOnlyGateway,
                                         false,
                                         metadataUpdater.getCluster(),
                                         null,
@@ -142,7 +147,8 @@ public CompletableFuture<SchemaInfo> getTableSchema(TablePath tablePath) {
         request.setTablePath()
                 .setDatabaseName(tablePath.getDatabaseName())
                 .setTableName(tablePath.getTableName());
-        return gateway.getTableSchema(request)
+        return readOnlyGateway
+                .getTableSchema(request)
                 .thenApply(
                         r ->
                                 new SchemaInfo(
@@ -157,7 +163,8 @@ public CompletableFuture<SchemaInfo> getTableSchema(TablePath tablePath, int sch
                 .setTablePath()
                 .setDatabaseName(tablePath.getDatabaseName())
                 .setTableName(tablePath.getTableName());
-        return gateway.getTableSchema(request)
+        return readOnlyGateway
+                .getTableSchema(request)
                 .thenApply(
                         r ->
                                 new SchemaInfo(
@@ -179,7 +186,8 @@ public CompletableFuture<Void> createDatabase(
     public CompletableFuture<DatabaseInfo> getDatabaseInfo(String databaseName) {
         GetDatabaseInfoRequest request = new GetDatabaseInfoRequest();
         request.setDatabaseName(databaseName);
-        return gateway.getDatabaseInfo(request)
+        return readOnlyGateway
+                .getDatabaseInfo(request)
                 .thenApply(
                         r ->
                                 new DatabaseInfo(
@@ -204,13 +212,14 @@ public CompletableFuture<Void> dropDatabase(
     public CompletableFuture<Boolean> databaseExists(String databaseName) {
         DatabaseExistsRequest request = new DatabaseExistsRequest();
         request.setDatabaseName(databaseName);
-        return gateway.databaseExists(request).thenApply(DatabaseExistsResponse::isExists);
+        return readOnlyGateway.databaseExists(request).thenApply(DatabaseExistsResponse::isExists);
     }
 
     @Override
     public CompletableFuture<List<String>> listDatabases() {
         ListDatabasesRequest request = new ListDatabasesRequest();
-        return gateway.listDatabases(request)
+        return readOnlyGateway
+                .listDatabases(request)
                 .thenApply(ListDatabasesResponse::getDatabaseNamesList);
     }
 
@@ -233,7 +242,8 @@ public CompletableFuture<TableInfo> getTableInfo(TablePath tablePath) {
         request.setTablePath()
                 .setDatabaseName(tablePath.getDatabaseName())
                 .setTableName(tablePath.getTableName());
-        return gateway.getTableInfo(request)
+        return readOnlyGateway
+                .getTableInfo(request)
                 .thenApply(
                         r ->
                                 TableInfo.of(
@@ -261,14 +271,14 @@ public CompletableFuture<Boolean> tableExists(TablePath tablePath) {
         request.setTablePath()
                 .setDatabaseName(tablePath.getDatabaseName())
                 .setTableName(tablePath.getTableName());
-        return gateway.tableExists(request).thenApply(TableExistsResponse::isExists);
+        return readOnlyGateway.tableExists(request).thenApply(TableExistsResponse::isExists);
     }
 
     @Override
     public CompletableFuture<List<String>> listTables(String databaseName) {
         ListTablesRequest request = new ListTablesRequest();
         request.setDatabaseName(databaseName);
-        return gateway.listTables(request).thenApply(ListTablesResponse::getTableNamesList);
+        return readOnlyGateway.listTables(request).thenApply(ListTablesResponse::getTableNamesList);
     }
 
     @Override
@@ -289,7 +299,8 @@ public CompletableFuture<List<PartitionInfo>> listPartitionInfos(
             PbPartitionSpec pbPartitionSpec = makePbPartitionSpec(partitionSpec);
             request.setPartialPartitionSpec(pbPartitionSpec);
         }
-        return gateway.listPartitionInfos(request)
+        return readOnlyGateway
+                .listPartitionInfos(request)
                 .thenApply(ClientRpcMessageUtils::toPartitionInfos);
     }
 
@@ -315,7 +326,8 @@ public CompletableFuture<KvSnapshots> getLatestKvSnapshots(TablePath tablePath)
         request.setTablePath()
                 .setDatabaseName(tablePath.getDatabaseName())
                 .setTableName(tablePath.getTableName());
-        return gateway.getLatestKvSnapshots(request)
+        return readOnlyGateway
+                .getLatestKvSnapshots(request)
                 .thenApply(ClientRpcMessageUtils::toKvSnapshots);
     }
 
@@ -328,7 +340,8 @@ public CompletableFuture<KvSnapshots> getLatestKvSnapshots(
                 .setDatabaseName(tablePath.getDatabaseName())
                 .setTableName(tablePath.getTableName());
         request.setPartitionName(partitionName);
-        return gateway.getLatestKvSnapshots(request)
+        return readOnlyGateway
+                .getLatestKvSnapshots(request)
                 .thenApply(ClientRpcMessageUtils::toKvSnapshots);
     }
 
@@ -342,7 +355,8 @@ public CompletableFuture<KvSnapshotMetadata> getKvSnapshotMetadata(
         request.setTableId(bucket.getTableId())
                 .setBucketId(bucket.getBucket())
                 .setSnapshotId(snapshotId);
-        return gateway.getKvSnapshotMetadata(request)
+        return readOnlyGateway
+                .getKvSnapshotMetadata(request)
                 .thenApply(ClientRpcMessageUtils::toKvSnapshotMetadata);
     }
 
@@ -353,7 +367,8 @@ public CompletableFuture<LakeSnapshot> getLatestLakeSnapshot(TablePath tablePath
                 .setDatabaseName(tablePath.getDatabaseName())
                 .setTableName(tablePath.getTableName());
 
-        return gateway.getLatestLakeSnapshot(request)
+        return readOnlyGateway
+                .getLatestLakeSnapshot(request)
                 .thenApply(ClientRpcMessageUtils::toLakeTableSnapshotInfo);
     }
 

fluss-client/src/main/java/com/alibaba/fluss/client/admin/FlussAdmin.java

@@ -34,8 +34,8 @@ public interface Lookuper {
      * Lookups certain row from the given lookup key.
      *
      * <p>The lookup key must be a primary key if the lookuper is a Primary Key Lookuper (created by
-     * {@code table.newLookuper().create()}), or be the prefix key if the lookuper is a Prefix Key
-     * Lookuper (created by {@code table.newLookuper().withLookupColumns(prefixKeys).create()}).
+     * {@code table.newLookup().createLookuper()}), or be the prefix key if the lookuper is a Prefix
+     * Key Lookuper (created by {@code table.newLookup().lookupBy(prefixKeys).createLookuper()}).
      *
      * @param lookupKey the lookup key.
      * @return the result of lookup.

fluss-client/src/main/java/com/alibaba/fluss/client/lookup/Lookuper.java

@@ -175,7 +175,7 @@ CompletedFetch poll() {
      *   <li>The thread was interrupted
      * </ol>
      *
-     * @param deadlineNanos the deadline time to wait util
+     * @param deadlineNanos the deadline time to wait until
      * @return false if the waiting time detectably elapsed before return from the method, else true
      */
     boolean awaitNotEmpty(long deadlineNanos) throws InterruptedException {