litiliu commented on code in PR #3639: URL: https://github.com/apache/fluss/pull/3639#discussion_r3585358280
########## fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergPartitionSpecValidator.java: ########## @@ -0,0 +1,45 @@ +/* + * 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. + */ + +package org.apache.fluss.lake.iceberg.tiering; + +import org.apache.fluss.exception.InvalidTableException; +import org.apache.fluss.lake.iceberg.utils.IcebergPartitionSpecUtils; +import org.apache.fluss.metadata.TableInfo; + +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Table; + +/** Validates that an Iceberg partition spec is compatible with Fluss lake tiering. */ +final class IcebergPartitionSpecValidator { + + private IcebergPartitionSpecValidator() {} + + static void validate(Table icebergTable, TableInfo tableInfo) { + PartitionSpec icebergSpec = icebergTable.spec(); + PartitionSpec expectedSpec = + IcebergPartitionSpecUtils.createPartitionSpec(tableInfo, icebergTable.schema()); + + if (!icebergSpec.compatibleWith(expectedSpec)) { Review Comment: Agreed. I changed the partition-spec comparison to compare the fields in order by `sourceId` and transform only, intentionally ignoring partition field names. Source column names, types, nullability, and order are still validated separately by the schema compatibility check. I also added `testAllowDifferentPartitionFieldNames` to verify that renaming only the partition fields does not block tiering. ########## fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java: ########## @@ -0,0 +1,104 @@ +/* + * 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. + */ + +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.fluss.metadata.TableInfo; + +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; + +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 +public final class IcebergPartitionSpecUtils { + + private IcebergPartitionSpecUtils() {} + + /** Creates an Iceberg partition spec from a Fluss table descriptor. */ + public static PartitionSpec createPartitionSpec( + TableDescriptor tableDescriptor, Schema icebergSchema) { + int bucketCount = + tableDescriptor + .getTableDistribution() + .flatMap(TableDescriptor.TableDistribution::getBucketCount) + .orElseThrow( + () -> + new IllegalArgumentException( + "Bucket count (bucket.num) must be set")); + return createPartitionSpec( + icebergSchema, + tableDescriptor.hasPrimaryKey(), + tableDescriptor.getBucketKeys(), + tableDescriptor.getPartitionKeys(), + bucketCount); + } + + /** Creates the expected Iceberg partition spec from resolved Fluss table metadata. */ + public static PartitionSpec createPartitionSpec(TableInfo tableInfo, Schema icebergSchema) { + return createPartitionSpec( + icebergSchema, + tableInfo.hasPrimaryKey(), + tableInfo.getBucketKeys(), + tableInfo.getPartitionKeys(), + tableInfo.getNumBuckets()); + } + + private static PartitionSpec createPartitionSpec( + Schema icebergSchema, + boolean isPrimaryKeyTable, + List<String> bucketKeys, + List<String> 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"); + } + + PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema); + for (String partitionKey : partitionKeys) { + if (!icebergSchema.findType(partitionKey).typeId().equals(STRING)) { Review Comment: If a source column is renamed in Iceberg, its schema no longer matches the Fluss schema because column names, types, nullability, and order are compared. The table will therefore fail validation with an `InvalidTableException`. I also added `testRejectRenamedIcebergColumn` to cover this case. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
