vinishjail97 commented on code in PR #935: URL: https://github.com/apache/incubator-xtable/pull/935#discussion_r4226929948
########## xtable-core/src/main/java/org/apache/xtable/index/HudiBackedIcebergSecondaryIndex.java: ########## @@ -0,0 +1,299 @@ +/* + * 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.Collections; +import java.util.List; +import java.util.Properties; +import java.util.Set; +import java.util.stream.Collectors; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; +import org.apache.spark.api.java.JavaPairRDD; +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.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.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.hadoop.HoodieHadoopStorage; + +import org.apache.iceberg.Table; + +import scala.Tuple2; + +import com.google.common.base.Preconditions; + +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; +import org.apache.xtable.model.sync.SyncResult; +import org.apache.xtable.model.sync.SyncStatusCode; + +/** + * 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 SourceTable sourceTable; + private final Properties targetTableProperties; + private final List<String> indexedColumns; + private final SparkSession sparkSession; + private final JavaSparkContext javaSparkContext; + private final HoodieSparkEngineContext engineContext; + + /** + * @param icebergTable the table to index + * @param sourceTable how XTable loads the table for a sync, because a loaded table does not + * expose the catalog it came from. A table of a catalog sets its catalog config, name and + * namespace, and a table without a catalog config is loaded from its location + * @param sparkSession the Spark session used to build and query the index + * @param targetTableProperties {@link HudiTargetConfig} properties for the sync. {@link + * HudiTargetConfig#SECONDARY_INDEX_COLUMNS} lists the columns to index, and other properties + * such as the record index file group counts are optional + */ + public HudiBackedIcebergSecondaryIndex( + Table icebergTable, + SourceTable sourceTable, + SparkSession sparkSession, + Properties targetTableProperties) { + Preconditions.checkArgument( + TableFormat.ICEBERG.equals(sourceTable.getFormatName()), + "Expected an Iceberg source table but got format %s", + sourceTable.getFormatName()); + this.tableLocation = icebergTable.location(); + this.dataPath = + TableFormatUtils.getTableDataLocation( + TableFormat.ICEBERG, tableLocation, icebergTable.properties()); + // the Hudi table lives under the data location, which Iceberg resolves from the table + // properties + this.sourceTable = sourceTable.toBuilder().dataPath(dataPath).build(); + this.targetTableProperties = new Properties(); + this.targetTableProperties.putAll(targetTableProperties); + this.targetTableProperties.setProperty( + HudiTargetConfig.EXECUTION_ENGINE, HudiTargetConfig.EXECUTION_ENGINE_SPARK); + // the secondary index is only available from Hudi table version 8 onwards + this.targetTableProperties.setProperty( + HudiTargetConfig.HUDI_TABLE_VERSION, String.valueOf(HoodieTableVersion.NINE.versionCode())); Review Comment: Done in ee95125b, see `getIndexTargetProperties`. ########## xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergPartitionValueConverter.java: ########## @@ -181,6 +182,28 @@ private InternalPartitionField getFromInternalPartitionFieldMap( Collectors.toMap(InternalPartitionField::getTransformType, Function.identity()))); } + /** + * Resolves the position of a partition field in the partition struct by name. The struct is Review Comment: `DataFile.partition()` is always a `PartitionData` in iceberg-core, so ee95125b removes the `IndexedRecord` branch and the dot escaping, and moves the tests to `PartitionData`. The lookup by name stays, because the source passes the current spec and not the file's spec (the TODO in `IcebergConversionSource`). ########## xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java: ########## @@ -35,22 +39,130 @@ public class HudiTargetConfig { */ public static final String HUDI_TABLE_VERSION = "xtable.hudi.target.table_version"; + /** + * Engine that writes the Hudi commit and the metadata table indexes. Supported values are {@code + * java} (default) and {@code spark}. The Spark engine reuses the active Spark session and + * distributes the index generation, which is recommended when the record index or a secondary + * index is enabled because those indexes read every registered data file. + */ + public static final String EXECUTION_ENGINE = "xtable.hudi.target.execution.engine"; + + public static final String EXECUTION_ENGINE_JAVA = "java"; + public static final String EXECUTION_ENGINE_SPARK = "spark"; + + /** + * Comma separated columns to build a Hudi secondary index on, one index per column. Setting it + * also enables the global record index, which the secondary index depends on. The indexes are + * stored in the Hudi metadata table under the target table path and require table version 9. The + * indexes are built when the Hudi table is created; a column added to the list of an existing + * table is not indexed yet. + */ + public static final String SECONDARY_INDEX_COLUMNS = "xtable.hudi.target.secondary.index.columns"; + + /** + * Lower and upper bound on the number of file groups of the record index. Both must be set + * together; when unset the Hudi defaults apply. + */ + public static final String RECORD_INDEX_MIN_FILEGROUP_COUNT = + "xtable.hudi.target.metadata.record.index.min.filegroup.count"; + + public static final String RECORD_INDEX_MAX_FILEGROUP_COUNT = + "xtable.hudi.target.metadata.record.index.max.filegroup.count"; + + /** Parallelism used when generating the secondary index records. */ + public static final String SECONDARY_INDEX_PARALLELISM = + "xtable.hudi.target.metadata.index.secondary.parallelism"; + static final HoodieTableVersion DEFAULT_TABLE_VERSION = HoodieTableVersion.SIX; HoodieTableVersion tableVersion; + boolean sparkEngine; + List<String> secondaryIndexColumns; + Optional<Integer> recordIndexMinFileGroupCount; + Optional<Integer> recordIndexMaxFileGroupCount; + Optional<Integer> secondaryIndexParallelism; public static HudiTargetConfig fromProperties(Properties properties) { + Properties targetProperties = properties == null ? new Properties() : properties; + HoodieTableVersion tableVersion = parseTableVersion(targetProperties); + boolean sparkEngine = parseSparkEngine(targetProperties); + List<String> secondaryIndexColumns = + Arrays.stream(targetProperties.getProperty(SECONDARY_INDEX_COLUMNS, "").split(",")) + .map(String::trim) + .filter(column -> !column.isEmpty()) + .distinct() + .collect(Collectors.toList()); + if (!secondaryIndexColumns.isEmpty() && tableVersion != HoodieTableVersion.NINE) { + throw new IllegalArgumentException( + String.format( + "%s requires Hudi target table version 9, set %s=9.", + SECONDARY_INDEX_COLUMNS, HUDI_TABLE_VERSION)); + } + if (secondaryIndexColumns.size() > 1 && !sparkEngine) { + // Hudi builds the index of the first column in the commit and the others with its indexer. + // In Hudi 1.2.1 the Java engine's JavaHoodieMetadataBulkInsertPartitioner writes an index + // built from existing files into a single file group, so lookups miss most keys. + throw new IllegalArgumentException( + String.format( + "More than one column in %s requires %s=%s.", + SECONDARY_INDEX_COLUMNS, EXECUTION_ENGINE, EXECUTION_ENGINE_SPARK)); + } + Optional<Integer> recordIndexMinFileGroupCount = + parsePositiveInt(targetProperties, RECORD_INDEX_MIN_FILEGROUP_COUNT); + Optional<Integer> recordIndexMaxFileGroupCount = + parsePositiveInt(targetProperties, RECORD_INDEX_MAX_FILEGROUP_COUNT); + if (recordIndexMinFileGroupCount.isPresent() != recordIndexMaxFileGroupCount.isPresent()) { + throw new IllegalArgumentException( + String.format( + "%s and %s must be set together.", + RECORD_INDEX_MIN_FILEGROUP_COUNT, RECORD_INDEX_MAX_FILEGROUP_COUNT)); + } + return new HudiTargetConfig( + tableVersion, + sparkEngine, + secondaryIndexColumns, + recordIndexMinFileGroupCount, + recordIndexMaxFileGroupCount, + parsePositiveInt(targetProperties, SECONDARY_INDEX_PARALLELISM)); + } + + private static HoodieTableVersion parseTableVersion(Properties properties) { HoodieTableVersion tableVersion = DEFAULT_TABLE_VERSION; - if (properties != null) { - String configured = properties.getProperty(HUDI_TABLE_VERSION); - if (configured != null && !configured.trim().isEmpty()) { - tableVersion = parseTableVersion(configured.trim()); - } + String configured = properties.getProperty(HUDI_TABLE_VERSION); + if (configured != null && !configured.trim().isEmpty()) { + tableVersion = parseTableVersion(configured.trim()); } if (tableVersion != HoodieTableVersion.SIX && tableVersion != HoodieTableVersion.NINE) { throw unsupportedTableVersion(String.valueOf(tableVersion.versionCode()), null); } - return new HudiTargetConfig(tableVersion); + return tableVersion; + } + + private static boolean parseSparkEngine(Properties properties) { + String engine = properties.getProperty(EXECUTION_ENGINE, EXECUTION_ENGINE_JAVA).trim(); Review Comment: Done in ee95125b with `HudiTargetConfig.ExecutionEngine`. ########## xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java: ########## @@ -35,22 +39,130 @@ public class HudiTargetConfig { */ public static final String HUDI_TABLE_VERSION = "xtable.hudi.target.table_version"; + /** + * Engine that writes the Hudi commit and the metadata table indexes. Supported values are {@code + * java} (default) and {@code spark}. The Spark engine reuses the active Spark session and + * distributes the index generation, which is recommended when the record index or a secondary + * index is enabled because those indexes read every registered data file. + */ + public static final String EXECUTION_ENGINE = "xtable.hudi.target.execution.engine"; + + public static final String EXECUTION_ENGINE_JAVA = "java"; + public static final String EXECUTION_ENGINE_SPARK = "spark"; Review Comment: Done in ee95125b with `HudiTargetConfig.ExecutionEngine`. ########## xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java: ########## @@ -482,6 +501,36 @@ private Optional<TableSyncMetadata> getMetadata( } } + /** + * Registers a definition for every secondary index partition that has none. Hudi updates every + * secondary index on each commit through its definition, but the commit that creates the metadata + * table rewrites the column-stats index definition from a stale view of the definitions, which + * drops the secondary index definition registered in the same commit. + */ + static void restoreSecondaryIndexDefinitions(HoodieTableMetaClient metaClient) { Review Comment: No, it was a workaround for apache/hudi#20245, where the commit that creates the metadata table drops the secondary index definition. With one column the commit names the column every time and Hudi registers the definition again, so ee95125b removes the workaround. A changed column now fails the sync before the commit. -- 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]
