fresh-borzoni commented on code in PR #3639: URL: https://github.com/apache/fluss/pull/3639#discussion_r3583903970
########## 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: nit: compatibleWith also compares partition field names. Two concerns: - simple updateSpec().renameField() blocks tiering, even though the write path never looks at field names - names are persisted in existing tables, so if createPartitionSpec ever changes its naming, every existing table fails validation after an upgrade Should we compare position + sourceId + transform instead? ########## 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: What happens if the column was renamed in iceberg instead of the spec changing? ########## fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java: ########## @@ -68,6 +68,7 @@ public IcebergLakeWriter( throws IOException { this.icebergCatalog = icebergCatalogProvider.get(); this.icebergTable = getTable(writerInitContext.tablePath()); + IcebergPartitionSpecValidator.validate(icebergTable, writerInitContext.tableInfo()); Review Comment: Should an incompatible table fail the whole tiering job (that's what happens when we throw here), or just that table? -- 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]
