diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java index 70886c9c83..1a255df55e 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java @@ -94,13 +94,6 @@ protected Catalog getIcebergCatalog() { @Override public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Context context) throws TableAlreadyExistException { - // validate at most one key field requirement - List keys = tableDescriptor.getBucketKeys(); - checkArgument( - keys.size() <= 1, - "Iceberg format supports at most one bucket key, but got: %s", - keys); - // convert Fluss table path to iceberg table boolean isPkTable = tableDescriptor.hasPrimaryKey(); TableIdentifier icebergId = toIcebergTableIdentifier(tablePath); diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java index 34c0217569..00cbec1733 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java @@ -178,8 +178,7 @@ private Function> createPartitionExtractor(Table tabl IntStream.range(0, partitionColCount).boxed().collect(Collectors.toList()); return task -> partitionFieldIndices.stream() - // since currently, only string partition is supported - .map(index -> task.partition().get(index, String.class)) + .map(index -> String.valueOf(task.partition().get(index, Object.class))) .collect(Collectors.toList()); } } diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java index a2e4c2a990..7c9af00b74 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java @@ -18,7 +18,6 @@ package org.apache.fluss.lake.iceberg.utils; import org.apache.fluss.annotation.Internal; -import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.metadata.TableDescriptor; import org.apache.iceberg.PartitionField; @@ -28,7 +27,6 @@ import java.util.List; import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; -import static org.apache.iceberg.types.Type.TypeID.STRING; /** Utilities for constructing the Iceberg partition spec used by Fluss lake tiering. */ @Internal @@ -61,11 +59,6 @@ private static PartitionSpec createPartitionSpec( List bucketKeys, List partitionKeys, int bucketCount) { - if (bucketKeys.size() > 1) { - throw new UnsupportedOperationException( - "Only one bucket key is supported for Iceberg at the moment"); - } - if (bucketKeys.isEmpty() && isPrimaryKeyTable) { throw new IllegalArgumentException( "Bucket key must be set for primary key Iceberg tables"); @@ -73,12 +66,6 @@ private static PartitionSpec createPartitionSpec( PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema); for (String partitionKey : partitionKeys) { - if (!icebergSchema.findType(partitionKey).typeId().equals(STRING)) { - throw new InvalidTableException( - String.format( - "Partition key only support string type for iceberg currently. Column `%s` is not string type.", - partitionKey)); - } builder.identity(partitionKey); } @@ -91,6 +78,9 @@ private static PartitionSpec createPartitionSpec( builder.identity(BUCKET_COLUMN_NAME); } } else { + // Use the first bucket key for Iceberg's bucket transform. For multi-key tables, + // this provides approximate scan pruning. The tiering writer explicitly assigns the + // correct bucket partition via PartitionKey regardless of how many keys are used. builder.bucket(bucketKeys.get(0), bucketCount); } return builder.build(); diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java index 868a6bfd47..00a8b40750 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java @@ -20,7 +20,6 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.InvalidAlterTableException; -import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.exception.TableAlreadyExistException; import org.apache.fluss.exception.TableNotExistException; import org.apache.fluss.lake.iceberg.testutils.IcebergTestUtils; @@ -46,7 +45,6 @@ import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.types.Types; -import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -267,14 +265,9 @@ void rejectsPrimaryKeyTableWithMultipleBucketKeys() { TablePath tablePath = TablePath.of(database, tableName); - assertThatThrownBy( - () -> - flussIcebergCatalog.createTable( - tablePath, - tableDescriptor, - new TestingLakeCatalogContext())) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Iceberg format supports at most one bucket key"); + // Multi-bucket-key tables should now create successfully + flussIcebergCatalog.createTable( + tablePath, tableDescriptor, new TestingLakeCatalogContext()); } @Test @@ -479,15 +472,9 @@ void rejectsLogTableWithMultipleBucketKeys() { TablePath tablePath = TablePath.of(database, tableName); - // Do not allow multiple bucket keys for log table - assertThatThrownBy( - () -> - flussIcebergCatalog.createTable( - tablePath, - tableDescriptor, - new TestingLakeCatalogContext())) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Iceberg format supports at most one bucket key"); + // Multi-bucket-key log tables should now create successfully + flussIcebergCatalog.createTable( + tablePath, tableDescriptor, new TestingLakeCatalogContext()); } @ParameterizedTest @@ -515,15 +502,9 @@ void testIllegalPartitionKeyType(boolean isPrimaryKeyTable) throws Exception { .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500)); tableDescriptor.partitionedBy(partitionKeys); - Assertions.assertThatThrownBy( - () -> - flussIcebergCatalog.createTable( - t1, - tableDescriptor.build(), - new TestingLakeCatalogContext())) - .isInstanceOf(InvalidTableException.class) - .hasMessage( - "Partition key only support string type for iceberg currently. Column `c1` is not string type."); + // Non-string partition keys should now be supported via identity transform + flussIcebergCatalog.createTable( + t1, tableDescriptor.build(), new TestingLakeCatalogContext()); } @Test