Merge branch 'main' into autoIncSyntax

Author: xx789633

.github/ISSUE_TEMPLATE/bug.yml

@@ -39,9 +39,8 @@ body:
       description: What Fluss version are you using?
       multiple: false
       options:
-        - "0.7.0 (latest release)"
-        - "0.6.0"
-        - "0.5.0"
+        - "0.8.0 (latest release)"
+        - "0.7.0"
         - "main (development)"
     validations:
       required: true

.github/workflows/docs-check.yaml

@@ -44,7 +44,7 @@ jobs:
         run: ./build_versioned_docs.sh
       - uses: actions/setup-node@v4
         with:
-          node-version: 18
+          node-version: 20
       - name: Install dependencies
         run: npm install
       - name: Test build website

LICENSE

@@ -401,3 +401,7 @@ LightProto
 ./fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufMessage.java
 ./fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufNumberField.java
 ./fluss-protogen/fluss-protogen-maven-plugin/src/main/java/org/apache/fluss/protogen/maven/plugin/ProtoGenMojo.java
+
+Apache Maven Wrapper
+./mvnw
+./mvnw.cmd

NOTICE

@@ -21,4 +21,19 @@ with the following copyright notice:
 LightProto
 Copyright 2020 Splunk Inc.
 
-----------------------------------------------------------
+----------------------------------------------------------
+
+This product contains code from the Apache Maven Wrapper Project:
+
+Apache Maven Wrapper
+Copyright 2013-2025 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+The original idea and initial implementation of the maven-wrapper module is derived
+from the Gradle Wrapper which was written originally by Hans Dockter and Adam Murdoch.
+Copyright 2007 the original author or authors.
+
+----------------------------------------------------------
+

README.md

@@ -1,3 +1,21 @@
+<!--
+ 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.
+-->
+
 <p align="center">
     <picture>
       <source media="(prefers-color-scheme: dark)" srcset="website/static/img/logo/svg/white_color_logo.svg">
@@ -43,15 +61,12 @@ 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
+- Java 11
 
 ```bash
 git clone https://github.com/apache/fluss.git
 cd fluss
-# in case of java 11
 ./mvnw clean package -DskipTests
-# or in case of java 8
-./mvnw clean package -DskipTests -Pjava8
 ```
 
 Apache Fluss is now installed in `build-target`. The build command uses Maven Wrapper (`mvnw`) which ensures the correct Maven version is used.

docker/quickstart-flink/README.md

@@ -1,3 +1,21 @@
+<!--
+ 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.
+-->
+
 # Fluss Quickstart Flink Docker
 
 This directory contains the Docker setup for Fluss Quickstart with Flink integration.

docker/quickstart-flink/prepare_build.sh

@@ -170,12 +170,12 @@ main() {
     # Iceberg Support
     log_info "Downloading Iceberg connector JARs..."
 
-    # Download iceberg-flink-runtime for Flink 1.20 (version 1.9.1)
+    # Download iceberg-flink-runtime for Flink 1.20 (version 1.10.0)
     download_jar \
-        "https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-runtime-1.20/1.9.1/iceberg-flink-runtime-1.20-1.9.1.jar" \
-        "./lib/iceberg-flink-runtime-1.20-1.9.1.jar" \
+        "https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-runtime-1.20/1.10.0/iceberg-flink-runtime-1.20-1.10.0.jar" \
+        "./lib/iceberg-flink-runtime-1.20-1.10.0.jar" \
         "" \
-        "iceberg-flink-runtime-1.20-1.9.1"
+        "iceberg-flink-runtime-1.20-1.10.0"
 
 
     # Prepare lake tiering JAR
@@ -201,7 +201,7 @@ verify_jars() {
         "flink-faker-0.5.3.jar"
         "hadoop-apache-3.3.5-2.jar"
         "paimon-flink-1.20-1.2.0.jar"
-        "iceberg-flink-runtime-1.20-1.9.1.jar"
+        "iceberg-flink-runtime-1.20-1.10.0.jar"
     )
 
     local opt_jars=(
@@ -250,7 +250,7 @@ show_summary() {
     echo "  ✓ Fluss Flink 1.20 connector"
     echo "  ✓ Fluss Lake Paimon connector"
     echo "  ✓ Fluss Lake Iceberg connector"
-    echo "  ✓ Iceberg Flink runtime 1.20 (v1.9.1)"
+    echo "  ✓ Iceberg Flink runtime 1.20 (v1.10.0)"
     echo "  ✓ Paimon Flink 1.20 (v1.2.0)"
     echo "  ✓ Hadoop Apache (v3.3.5-2)"
     echo "  ✓ Flink Faker (v0.5.3)"

fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java

@@ -33,7 +33,6 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.fs.FileSystem;
-import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.metrics.registry.MetricRegistry;
 import org.apache.fluss.rpc.GatewayClientProxy;
@@ -61,6 +60,7 @@ public final class FlussConnection implements Connection {
     private volatile LookupClient lookupClient;
     private volatile RemoteFileDownloader remoteFileDownloader;
     private volatile SecurityTokenManager securityTokenManager;
+    private volatile Admin admin;
 
     FlussConnection(Configuration conf) {
         this(conf, MetricRegistry.create(conf, null));
@@ -93,19 +93,15 @@ public Configuration getConfiguration() {
 
     @Override
     public Admin getAdmin() {
-        return new FlussAdmin(rpcClient, metadataUpdater);
+        return getOrCreateAdmin();
     }
 
     @Override
     public Table getTable(TablePath tablePath) {
-        // force to update the table info from server to avoid stale data in cache
+        // force to update the table info from server to avoid stale data in cache.
         metadataUpdater.updateTableOrPartitionMetadata(tablePath, null);
-        TableInfo tableInfo = metadataUpdater.getTableInfoOrElseThrow(tablePath);
-        return new FlussTable(this, tablePath, tableInfo);
-    }
-
-    public RpcClient getRpcClient() {
-        return rpcClient;
+        Admin admin = getOrCreateAdmin();
+        return new FlussTable(this, tablePath, admin.getTableInfo(tablePath).join());
     }
 
     public MetadataUpdater getMetadataUpdater() {
@@ -140,6 +136,17 @@ public LookupClient getOrCreateLookupClient() {
         return lookupClient;
     }
 
+    public Admin getOrCreateAdmin() {
+        if (admin == null) {
+            synchronized (this) {
+                if (admin == null) {
+                    admin = new FlussAdmin(rpcClient, metadataUpdater);
+                }
+            }
+        }
+        return admin;
+    }
+
     public RemoteFileDownloader getOrCreateRemoteFileDownloader() {
         if (remoteFileDownloader == null) {
             synchronized (this) {

fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java

@@ -88,8 +88,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
 
+import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeAlterTableRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeCreatePartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeDropPartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeListOffsetsRequest;
@@ -248,18 +248,8 @@ public CompletableFuture<Void> createTable(
     public CompletableFuture<Void> alterTable(
             TablePath tablePath, List<TableChange> tableChanges, boolean ignoreIfNotExists) {
         tablePath.validate();
-        AlterTableRequest request = new AlterTableRequest();
-
-        List<PbAlterConfig> pbFlussTableChanges =
-                tableChanges.stream()
-                        .map(ClientRpcMessageUtils::toPbAlterConfigs)
-                        .collect(Collectors.toList());
-
-        request.addAllConfigChanges(pbFlussTableChanges)
-                .setIgnoreIfNotExists(ignoreIfNotExists)
-                .setTablePath()
-                .setDatabaseName(tablePath.getDatabaseName())
-                .setTableName(tablePath.getTableName());
+        AlterTableRequest request =
+                makeAlterTableRequest(tablePath, tableChanges, ignoreIfNotExists);
         return gateway.alterTable(request).thenApply(r -> null);
     }
 
@@ -420,15 +410,16 @@ private ListOffsetsResult listOffsets(
             OffsetSpec offsetSpec) {
         Long partitionId = null;
         metadataUpdater.updateTableOrPartitionMetadata(physicalTablePath.getTablePath(), null);
-        long tableId = metadataUpdater.getTableId(physicalTablePath.getTablePath());
+        TableInfo tableInfo = getTableInfo(physicalTablePath.getTablePath()).join();
+
         // if partition name is not null, we need to check and update partition metadata
         if (physicalTablePath.getPartitionName() != null) {
             metadataUpdater.updatePhysicalTableMetadata(Collections.singleton(physicalTablePath));
             partitionId = metadataUpdater.getPartitionIdOrElseThrow(physicalTablePath);
         }
         Map<Integer, ListOffsetsRequest> requestMap =
                 prepareListOffsetsRequests(
-                        metadataUpdater, tableId, partitionId, buckets, offsetSpec);
+                        metadataUpdater, tableInfo.getTableId(), partitionId, buckets, offsetSpec);
         Map<Integer, CompletableFuture<Long>> bucketToOffsetMap = MapUtils.newConcurrentHashMap();
         for (int bucket : buckets) {
             bucketToOffsetMap.put(bucket, new CompletableFuture<>());

fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java

@@ -24,8 +24,9 @@
 
 import java.math.BigDecimal;
 import java.util.Arrays;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -43,7 +44,7 @@ final class ConverterCommons {
     static final Map<DataTypeRoot, Set<Class<?>>> SUPPORTED_TYPES = createSupportedTypes();
 
     private static Map<DataTypeRoot, Set<Class<?>>> createSupportedTypes() {
-        Map<DataTypeRoot, Set<Class<?>>> map = new HashMap<>();
+        Map<DataTypeRoot, Set<Class<?>>> map = new EnumMap<>(DataTypeRoot.class);
         map.put(DataTypeRoot.BOOLEAN, setOf(Boolean.class));
         map.put(DataTypeRoot.TINYINT, setOf(Byte.class));
         map.put(DataTypeRoot.SMALLINT, setOf(Short.class));
@@ -67,15 +68,15 @@ private static Map<DataTypeRoot, Set<Class<?>>> createSupportedTypes() {
 
     static void validatePojoMatchesTable(PojoType<?> pojoType, RowType tableSchema) {
         Set<String> pojoNames = pojoType.getProperties().keySet();
-        Set<String> tableNames = new HashSet<>(tableSchema.getFieldNames());
-        if (!pojoNames.equals(tableNames)) {
+        List<String> fieldNames = tableSchema.getFieldNames();
+        if (!pojoNames.containsAll(fieldNames)) {
             throw new IllegalArgumentException(
                     String.format(
                             "POJO fields %s must exactly match table schema fields %s.",
-                            pojoNames, tableNames));
+                            pojoNames, fieldNames));
         }
         for (int i = 0; i < tableSchema.getFieldCount(); i++) {
-            String name = tableSchema.getFieldNames().get(i);
+            String name = fieldNames.get(i);
             DataType dt = tableSchema.getTypeAt(i);
             PojoType.Property prop = pojoType.getProperty(name);
             validateCompatibility(dt, prop);