vinishjail97 commented on code in PR #935: URL: https://github.com/apache/incubator-xtable/pull/935#discussion_r4203877784
########## xtable-core/src/main/java/org/apache/xtable/index/IcebergPartitionConverter.java: ########## @@ -0,0 +1,159 @@ +/* + * 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.xtable.index; + +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.util.List; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +import org.apache.avro.generic.GenericData; +import org.apache.avro.util.Utf8; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Decimal; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.unsafe.types.UTF8String; + +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ByteBuffers; + +/** + * Converts Iceberg partition values to the Spark {@link InternalRow} representation Iceberg's own + * Spark reader exposes through the {@code _partition} metadata column. + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class IcebergPartitionConverter { + + /** + * Parses a partition path (for example {@code year=2024/month=01}) with the given spec and + * converts the values to an {@link InternalRow}. + * + * @return the partition row, or null when the table is unpartitioned + */ + public static InternalRow convertPartitionToInternalRow( + String partitionPath, Types.StructType partitionType, PartitionSpec spec) { + if (partitionPath == null || partitionPath.isEmpty() || partitionType.fields().isEmpty()) { + return null; + } + return convertStructLikeToInternalRow(DataFiles.data(spec, partitionPath), partitionType); Review Comment: [P1] Decode actual Iceberg partition values DataFiles.data does not reverse PartitionSpec.partitionToPath. With Iceberg 1.9.2 paths, the checks returned 2024 instead of 54 for year 2024, threw NumberFormatException for month 2024-01 and a null integer partition, and returned a%2Fb instead of a/b. These cases use one partition spec. Using DataFile partition metadata or decoding that handles transforms and path encoding would preserve the correct values. Tests should cover these round-trip cases because incorrect _partition values can break merge and delete joins. ########## xtable-core/src/main/java/org/apache/xtable/index/HudiBackedIcebergSecondaryIndex.java: ########## @@ -0,0 +1,237 @@ +/* + * 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.xtable.index; + +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX; +import static org.apache.spark.sql.functions.col; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Properties; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Encoders; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.catalyst.CatalystTypeConverters; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.data.HoodiePairData; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.data.HoodieJavaPairRDD; +import org.apache.hudi.data.HoodieJavaRDD; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.hadoop.HoodieHadoopStorage; + +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Partitioning; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Types; + +import scala.Function1; +import scala.Tuple2; + +import org.apache.xtable.catalog.TableFormatUtils; +import org.apache.xtable.conversion.ConversionConfig; +import org.apache.xtable.conversion.ConversionController; +import org.apache.xtable.conversion.SourceTable; +import org.apache.xtable.conversion.TargetTable; +import org.apache.xtable.hudi.HudiTargetConfig; +import org.apache.xtable.iceberg.IcebergConversionSourceProvider; +import org.apache.xtable.model.storage.TableFormat; + +/** + * Secondary index for an Iceberg table backed by the Hudi metadata table. The index is built by + * syncing the Iceberg table to Hudi with the record index and a secondary index enabled; the Hudi + * metadata lives under {@code <data location>/.hoodie/metadata}. Because the Iceberg data files + * carry no Hudi record keys, the record key of a row is {@code <file path relative to the data + * location>_<row position>}, which a lookup resolves back to a file path and row position. + */ +@Log4j2 +public class HudiBackedIcebergSecondaryIndex implements Index<Table> { + private final String tableLocation; + private final String dataPath; + private final Properties targetTableProperties; + private final SparkSession sparkSession; + private final JavaSparkContext javaSparkContext; + private final HoodieSparkEngineContext engineContext; + + /** + * @param icebergTable the table to index + * @param sparkSession the Spark session used to build and query the index + * @param targetTableProperties additional {@link HudiTargetConfig} properties for the sync, for + * example the record index file group counts + */ + public HudiBackedIcebergSecondaryIndex( + Table icebergTable, SparkSession sparkSession, Properties targetTableProperties) { + this.tableLocation = icebergTable.location(); + this.dataPath = + TableFormatUtils.getTableDataLocation( + TableFormat.ICEBERG, tableLocation, icebergTable.properties()); + this.targetTableProperties = targetTableProperties; + this.sparkSession = sparkSession; + this.javaSparkContext = JavaSparkContext.fromSparkContext(sparkSession.sparkContext()); + this.engineContext = new HoodieSparkEngineContext(javaSparkContext); + } + + @Override + public boolean doesIndexExist(String columnName) { + return HoodieTableMetadataUtil.metadataPartitionExists( + dataPath, engineContext, PARTITION_NAME_SECONDARY_INDEX_PREFIX + columnName); + } + + @Override + public void syncIndex(Table icebergTable, String columnName) { + Configuration configuration = javaSparkContext.hadoopConfiguration(); + IcebergConversionSourceProvider sourceProvider = new IcebergConversionSourceProvider(); + sourceProvider.init(configuration); + new ConversionController(configuration) + .sync(getConversionConfig(icebergTable, columnName), sourceProvider); Review Comment: [P1] Propagate failed sync results TableFormatSync catches target exceptions and returns SyncStatusCode.ERROR. ConversionController.sync returns that result, but syncIndex discards it and logs success. A missing_column reproduction failed inside Hudi while syncIndex returned normally. An existing index can therefore remain stale without notifying the caller. Checking the result and propagating the failure would preserve the sync contract. Tests should cover failed initial syncs and failed updates. ########## xtable-core/src/main/java/org/apache/xtable/index/HudiBackedIcebergSecondaryIndex.java: ########## @@ -0,0 +1,237 @@ +/* + * 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.xtable.index; + +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX; +import static org.apache.spark.sql.functions.col; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Properties; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Encoders; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.catalyst.CatalystTypeConverters; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.data.HoodiePairData; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.data.HoodieJavaPairRDD; +import org.apache.hudi.data.HoodieJavaRDD; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.hadoop.HoodieHadoopStorage; + +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Partitioning; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Types; + +import scala.Function1; +import scala.Tuple2; + +import org.apache.xtable.catalog.TableFormatUtils; +import org.apache.xtable.conversion.ConversionConfig; +import org.apache.xtable.conversion.ConversionController; +import org.apache.xtable.conversion.SourceTable; +import org.apache.xtable.conversion.TargetTable; +import org.apache.xtable.hudi.HudiTargetConfig; +import org.apache.xtable.iceberg.IcebergConversionSourceProvider; +import org.apache.xtable.model.storage.TableFormat; + +/** + * Secondary index for an Iceberg table backed by the Hudi metadata table. The index is built by + * syncing the Iceberg table to Hudi with the record index and a secondary index enabled; the Hudi + * metadata lives under {@code <data location>/.hoodie/metadata}. Because the Iceberg data files + * carry no Hudi record keys, the record key of a row is {@code <file path relative to the data + * location>_<row position>}, which a lookup resolves back to a file path and row position. + */ +@Log4j2 +public class HudiBackedIcebergSecondaryIndex implements Index<Table> { + private final String tableLocation; + private final String dataPath; + private final Properties targetTableProperties; + private final SparkSession sparkSession; + private final JavaSparkContext javaSparkContext; + private final HoodieSparkEngineContext engineContext; + + /** + * @param icebergTable the table to index + * @param sparkSession the Spark session used to build and query the index + * @param targetTableProperties additional {@link HudiTargetConfig} properties for the sync, for + * example the record index file group counts + */ + public HudiBackedIcebergSecondaryIndex( + Table icebergTable, SparkSession sparkSession, Properties targetTableProperties) { + this.tableLocation = icebergTable.location(); + this.dataPath = + TableFormatUtils.getTableDataLocation( + TableFormat.ICEBERG, tableLocation, icebergTable.properties()); + this.targetTableProperties = targetTableProperties; + this.sparkSession = sparkSession; + this.javaSparkContext = JavaSparkContext.fromSparkContext(sparkSession.sparkContext()); + this.engineContext = new HoodieSparkEngineContext(javaSparkContext); + } + + @Override + public boolean doesIndexExist(String columnName) { + return HoodieTableMetadataUtil.metadataPartitionExists( + dataPath, engineContext, PARTITION_NAME_SECONDARY_INDEX_PREFIX + columnName); + } + + @Override + public void syncIndex(Table icebergTable, String columnName) { + Configuration configuration = javaSparkContext.hadoopConfiguration(); + IcebergConversionSourceProvider sourceProvider = new IcebergConversionSourceProvider(); + sourceProvider.init(configuration); + new ConversionController(configuration) + .sync(getConversionConfig(icebergTable, columnName), sourceProvider); + log.info("Synced secondary index for column {} of table {}", columnName, tableLocation); + } + + private ConversionConfig getConversionConfig(Table icebergTable, String columnName) { + SourceTable sourceTable = + SourceTable.builder() + .name(icebergTable.name()) + .basePath(tableLocation) + .dataPath(dataPath) + .formatName(TableFormat.ICEBERG) + .build(); Review Comment: [P2] Preserve catalog context SourceTable contains the table location but no catalog configuration. IcebergConversionSource therefore falls back to HadoopTables.load instead of using the supplied Table. A valid InMemoryCatalog table reproduced NoSuchTableException when syncIndex reloaded its location. The existing tests use Hadoop tables and do not cover this case. Using the supplied Table would preserve its catalog context. Alternatively, a Hadoop-table-only contract needs explicit validation and documentation. ########## xtable-core/src/main/java/org/apache/xtable/index/HudiBackedIcebergSecondaryIndex.java: ########## @@ -0,0 +1,237 @@ +/* + * 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.xtable.index; + +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX; +import static org.apache.spark.sql.functions.col; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Properties; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Encoders; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.catalyst.CatalystTypeConverters; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.data.HoodiePairData; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.data.HoodieJavaPairRDD; +import org.apache.hudi.data.HoodieJavaRDD; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.hadoop.HoodieHadoopStorage; + +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Partitioning; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Types; + +import scala.Function1; +import scala.Tuple2; + +import org.apache.xtable.catalog.TableFormatUtils; +import org.apache.xtable.conversion.ConversionConfig; +import org.apache.xtable.conversion.ConversionController; +import org.apache.xtable.conversion.SourceTable; +import org.apache.xtable.conversion.TargetTable; +import org.apache.xtable.hudi.HudiTargetConfig; +import org.apache.xtable.iceberg.IcebergConversionSourceProvider; +import org.apache.xtable.model.storage.TableFormat; + +/** + * Secondary index for an Iceberg table backed by the Hudi metadata table. The index is built by + * syncing the Iceberg table to Hudi with the record index and a secondary index enabled; the Hudi + * metadata lives under {@code <data location>/.hoodie/metadata}. Because the Iceberg data files + * carry no Hudi record keys, the record key of a row is {@code <file path relative to the data + * location>_<row position>}, which a lookup resolves back to a file path and row position. + */ +@Log4j2 +public class HudiBackedIcebergSecondaryIndex implements Index<Table> { + private final String tableLocation; + private final String dataPath; + private final Properties targetTableProperties; + private final SparkSession sparkSession; + private final JavaSparkContext javaSparkContext; + private final HoodieSparkEngineContext engineContext; + + /** + * @param icebergTable the table to index + * @param sparkSession the Spark session used to build and query the index + * @param targetTableProperties additional {@link HudiTargetConfig} properties for the sync, for + * example the record index file group counts + */ + public HudiBackedIcebergSecondaryIndex( + Table icebergTable, SparkSession sparkSession, Properties targetTableProperties) { + this.tableLocation = icebergTable.location(); + this.dataPath = + TableFormatUtils.getTableDataLocation( + TableFormat.ICEBERG, tableLocation, icebergTable.properties()); + this.targetTableProperties = targetTableProperties; + this.sparkSession = sparkSession; + this.javaSparkContext = JavaSparkContext.fromSparkContext(sparkSession.sparkContext()); + this.engineContext = new HoodieSparkEngineContext(javaSparkContext); + } + + @Override + public boolean doesIndexExist(String columnName) { + return HoodieTableMetadataUtil.metadataPartitionExists( + dataPath, engineContext, PARTITION_NAME_SECONDARY_INDEX_PREFIX + columnName); + } + + @Override + public void syncIndex(Table icebergTable, String columnName) { + Configuration configuration = javaSparkContext.hadoopConfiguration(); + IcebergConversionSourceProvider sourceProvider = new IcebergConversionSourceProvider(); + sourceProvider.init(configuration); + new ConversionController(configuration) + .sync(getConversionConfig(icebergTable, columnName), sourceProvider); + log.info("Synced secondary index for column {} of table {}", columnName, tableLocation); + } + + private ConversionConfig getConversionConfig(Table icebergTable, String columnName) { + SourceTable sourceTable = + SourceTable.builder() + .name(icebergTable.name()) + .basePath(tableLocation) + .dataPath(dataPath) + .formatName(TableFormat.ICEBERG) + .build(); + Properties mergedProperties = new Properties(); + mergedProperties.putAll(targetTableProperties); + mergedProperties.setProperty(HudiTargetConfig.SECONDARY_INDEX_COLUMN, columnName); + mergedProperties.setProperty( + HudiTargetConfig.EXECUTION_ENGINE, HudiTargetConfig.EXECUTION_ENGINE_SPARK); + // the secondary index is only available from Hudi table version 8 onwards + mergedProperties.setProperty( + HudiTargetConfig.HUDI_TABLE_VERSION, String.valueOf(HoodieTableVersion.NINE.versionCode())); + TargetTable targetTable = + TargetTable.builder() + .name(icebergTable.name()) + .basePath(dataPath) + .formatName(TableFormat.HUDI) + .additionalProperties(mergedProperties) + .build(); + return ConversionConfig.builder() + .sourceTable(sourceTable) + .targetTables(Collections.singletonList(targetTable)) + .build(); + } + + @Override + public Dataset<Row> lookup(Table icebergTable, Dataset<Row> keys, String columnName) { + Types.StructType partitionType = Partitioning.partitionType(icebergTable); + PartitionSpec spec = icebergTable.spec(); + StructType partitionSparkType = IcebergPartitionConverter.toSparkType(partitionType); + StructType resultSchema = + new StructType() + .add(columnName, DataTypes.StringType, false) + .add(FILE_COLUMN, DataTypes.StringType, false) + .add(POSITION_COLUMN, DataTypes.LongType, false) + .add(PARTITION_COLUMN, partitionSparkType, true); + // the secondary index stores the values of the indexed column as strings + JavaRDD<String> secondaryKeys = + keys.where(col(columnName).isNotNull()) + .select(col(columnName).cast(DataTypes.StringType)) + .as(Encoders.STRING()) + .toJavaRDD(); + HoodieStorage storage = + new HoodieHadoopStorage(dataPath, javaSparkContext.hadoopConfiguration()); + HoodieTableMetaClient metaClient = + HoodieTableMetaClient.builder().setStorage(storage).setBasePath(dataPath).build(); + HoodieMetadataConfig metadataConfig = + HoodieMetadataConfig.newBuilder().enable(true).withSecondaryIndexEnabled(true).build(); + try (HoodieBackedTableMetadata tableMetadata = + (HoodieBackedTableMetadata) + metaClient + .getTableFormat() + .getMetadataFactory() + .create(engineContext, storage, metadataConfig, dataPath)) { + HoodiePairData<String, String> recordKeysBySecondaryKey = + tableMetadata.readSecondaryIndexDataTableRecordKeysWithKeys( + HoodieJavaRDD.of(secondaryKeys), PARTITION_NAME_SECONDARY_INDEX_PREFIX + columnName); + // capture the field so the closure does not serialize the index instance + String dataPath = this.dataPath; + JavaRDD<Row> lookupResults = + HoodieJavaPairRDD.getJavaPairRDD(recordKeysBySecondaryKey) Review Comment: [P2] Handle list-backed lookup results Hudi 1.2.1 returns HoodieListPairData for empty secondary keys, while getJavaPairRDD casts directly to HoodieJavaPairRDD. Empty and null-only DataFrames both reproduced ClassCastException. These requests should return an empty DataFrame with resultSchema. Handling the list-backed result type would also cover Hudi's path for one file group. Tests should cover empty and null-only requests. -- 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]
