[lake/iceberg] thrown Exception when partition columns are of non-String type

Author: wangjunbo

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InvalidTableException;
 import org.apache.fluss.exception.TableAlreadyExistException;
 import org.apache.fluss.exception.TableNotExistException;
 import org.apache.fluss.lake.iceberg.utils.IcebergCatalogUtils;
@@ -51,6 +52,7 @@
 import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
 import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
 import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
+import static org.apache.iceberg.types.Type.TypeID.STRING;
 
 /** An Iceberg implementation of {@link LakeCatalog}. */
 public class IcebergLakeCatalog implements LakeCatalog {
@@ -214,6 +216,14 @@ private PartitionSpec createPartitionSpec(
         List<String> partitionKeys = tableDescriptor.getPartitionKeys();
         // always set identity partition with partition key
         for (String partitionKey : partitionKeys) {
+            if (!icebergSchema.findType(partitionKey).typeId().equals(STRING)) {
+                // TODO: Support other types of partition column
+                throw new InvalidTableException(
+                        String.format(
+                                "Iceberg partition column only support string type, "
+                                        + "%s is not string type.",
+                                partitionKey));
+            }
             builder.identity(partitionKey);
         }
 

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringITCase.java

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.config.AutoPartitionTimeUnit;
 import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.exception.InvalidTableException;
 import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase;
 import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.TableBucket;
@@ -39,6 +40,10 @@
 import org.apache.iceberg.data.Record;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
@@ -53,10 +58,13 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Stream;
 
 import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.testutils.DataTestUtils.row;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** The ITCase for tiering into iceberg. */
 class IcebergTieringITCase extends FlinkIcebergTieringTestBase {
@@ -86,7 +94,7 @@ class IcebergTieringITCase extends FlinkIcebergTieringTestBase {
                     .column("f_time", DataTypes.TIME())
                     .column("f_char", DataTypes.CHAR(3))
                     .column("f_bytes", DataTypes.BYTES())
-                    .primaryKey("f_int")
+                    .primaryKey("f_date", "f_int")
                     .build();
 
     private static final Schema logSchema =
@@ -278,6 +286,44 @@ void testTiering() throws Exception {
         }
     }
 
+    private static Stream<Arguments> tieringAllTypesWriteArgs() {
+        return Stream.of(Arguments.of(true), Arguments.of(false));
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void testTieringForAllTypes(boolean isPrimaryKeyTable) throws Exception {
+        // create a table, write some records and wait until snapshot finished
+        TablePath t1 =
+                TablePath.of(
+                        DEFAULT_DB,
+                        isPrimaryKeyTable ? "pkTableForAllTypes" : "logTableForAllTypes");
+        Schema.Builder builder =
+                Schema.newBuilder()
+                        .column("c0", DataTypes.STRING())
+                        .column("c1", DataTypes.BOOLEAN());
+        if (isPrimaryKeyTable) {
+            builder.primaryKey("c0", "c1");
+        }
+        List<String> partitionKeys = List.of("c1");
+        TableDescriptor.Builder tableDescriptor =
+                TableDescriptor.builder()
+                        .schema(builder.build())
+                        .distributedBy(1, "c0")
+                        .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")
+                        .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500));
+        tableDescriptor.partitionedBy(partitionKeys);
+        tableDescriptor.customProperties(Collections.emptyMap());
+        tableDescriptor.properties(Collections.emptyMap());
+
+        assertThatThrownBy(() -> createTable(t1, tableDescriptor.build()))
+                .isInstanceOf(ExecutionException.class)
+                .rootCause()
+                .isInstanceOf(InvalidTableException.class)
+                .hasMessage(
+                        "Iceberg partition column only support string type, c1 is not string type.");
+    }
+
     private void checkDataInIcebergPrimaryKeyTable(
             TablePath tablePath, List<InternalRow> expectedRows) throws Exception {
         Iterator<Record> acturalIterator = getIcebergRecords(tablePath).iterator();