vinishjail97 commented on code in PR #935: URL: https://github.com/apache/incubator-xtable/pull/935#discussion_r4252593979
########## xtable-core/src/main/java/org/apache/xtable/index/HudiBackedSecondaryIndexReader.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.stream.Collectors; + +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.DataType; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.DateType; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.types.TimestampNTZType; +import org.apache.spark.sql.types.TimestampType; + +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.fs.FSUtils; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.data.HoodieJavaPairRDD; +import org.apache.hudi.data.HoodieJavaRDD; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; + +import scala.Tuple2; + +import org.apache.xtable.conversion.ConversionSourceProvider; +import org.apache.xtable.conversion.SourceTable; +import org.apache.xtable.model.schema.InternalField; +import org.apache.xtable.model.schema.InternalSchema; +import org.apache.xtable.model.schema.InternalType; +import org.apache.xtable.schema.SparkSchemaExtractor; + +/** + * Reads the secondary index that {@link HudiBackedSecondaryIndexWriter} builds for a table of any + * source format, and returns the file and row position of each match. The record key of a row in + * the index is {@code <file path relative to the data path>_<row position>}, which the lookup + * resolves back to a file path and row position. + * + * <p>Hudi keys the index by the string of Spark's internal value of the column: the day count of a + * DATE, the microseconds of a TIMESTAMP (with or without time zone) and the plain string of other + * types. A lookup renders its keys the same way and returns them in the type of the column. + */ +public class HudiBackedSecondaryIndexReader + implements IndexReader<SourceTable, Dataset<Row>, Dataset<Row>> { + /** + * Lookup result column with the full path of the data file that holds the row. The name matches + * Iceberg's {@code _file} metadata column, so the result joins to an Iceberg table directly. + */ + public static final String FILE_COLUMN = "_file"; + + /** + * Lookup result column with the zero based position of the row within the file. The name matches + * Iceberg's {@code _pos} metadata column. + */ + public static final String POSITION_COLUMN = "_pos"; + + // the key as Hudi renders it in the secondary index + private static final String SECONDARY_KEY_COLUMN = "secondary_key"; + + private static final String NTZ_EPOCH = "TIMESTAMP_NTZ'1970-01-01 00:00:00'"; + + private final String tableLocation; + private final String dataPath; + private final SourceTable sourceTable; + private final ConversionSourceProvider<?> sourceProvider; + private final SparkSession sparkSession; + private final JavaSparkContext javaSparkContext; + private final HoodieSparkEngineContext engineContext; + + /** + * @param sourceTable the indexed table, as given to {@link HudiBackedSecondaryIndexWriter} + * @param sourceProvider the provider of the conversion source for the format of the table, which + * reads the type of the looked up column. The lookup initializes it with the Hadoop + * configuration of the Spark session + * @param sparkSession the Spark session used to read the index + */ + public HudiBackedSecondaryIndexReader( + SourceTable sourceTable, + ConversionSourceProvider<?> sourceProvider, + SparkSession sparkSession) { + this.sourceTable = sourceTable; + this.tableLocation = sourceTable.getBasePath(); + this.dataPath = sourceTable.getDataPath(); + this.sparkSession = sparkSession; + this.javaSparkContext = JavaSparkContext.fromSparkContext(sparkSession.sparkContext()); + this.engineContext = new HoodieSparkEngineContext(javaSparkContext); + this.sourceProvider = sourceProvider; + sourceProvider.init(javaSparkContext.hadoopConfiguration()); + } + + /** Returns the Spark type of the column, which decides how Hudi renders its values. */ + private DataType getKeyType(String column) { + InternalField field = + HudiBackedSecondaryIndexUtils.getIndexableField( + HudiBackedSecondaryIndexUtils.readSchema(sourceProvider, sourceTable), column); + InternalSchema keySchema = + InternalSchema.builder() + .name("key") + .dataType(InternalType.RECORD) + .fields(Collections.singletonList(field)) + .build(); + return SparkSchemaExtractor.getInstance().fromInternalSchema(keySchema).fields()[0].dataType(); + } + + /** Renders a key of the given type the way Hudi renders the value in the secondary index. */ + private static String toSecondaryKey(DataType keyType, String key) { + if (keyType instanceof DateType) { + return "CAST(unix_date(" + key + ") AS STRING)"; + } + if (keyType instanceof TimestampType) { + return "CAST(unix_micros(" + key + ") AS STRING)"; + } + if (keyType instanceof TimestampNTZType) { + // unix_micros takes only a TIMESTAMP, and a cast to one would apply the session time zone + return "CAST(timestampdiff(MICROSECOND, " + NTZ_EPOCH + ", " + key + ") AS STRING)"; + } + return "CAST(" + key + " AS STRING)"; + } + + /** Reverses {@link #toSecondaryKey} to return a key in the type of the indexed column. */ + private static String fromSecondaryKey(DataType keyType, String key) { + if (keyType instanceof DateType) { + return "date_from_unix_date(CAST(" + key + " AS INT))"; + } + if (keyType instanceof TimestampType) { + return "timestamp_micros(CAST(" + key + " AS BIGINT))"; + } + if (keyType instanceof TimestampNTZType) { + // timestampadd takes an INT count, so add the exact interval between two instants instead + return NTZ_EPOCH + + " + (timestamp_micros(CAST(" + + key + + " AS BIGINT)) - TIMESTAMP'1970-01-01 00:00:00+00:00')"; + } + return "CAST(" + key + " AS " + keyType.sql() + ")"; + } + + @Override + public boolean doesIndexExist(String columnName) { + return HudiBackedSecondaryIndexUtils.getIndexedColumns( + dataPath, javaSparkContext.hadoopConfiguration()) + .contains(columnName); + } + + @Override + public Dataset<Row> lookup(SourceTable table, Dataset<Row> keys, String columnName) { + HudiBackedSecondaryIndexUtils.checkIndexedTable(tableLocation, table); + HoodieTableMetaClient metaClient = + HudiBackedSecondaryIndexUtils.getMetaClient( + dataPath, javaSparkContext.hadoopConfiguration()); + if (!HudiBackedSecondaryIndexUtils.getIndexedColumns(metaClient).contains(columnName)) { + throw new IllegalStateException( + "The secondary index of column " + + columnName + + " of table " + + tableLocation + + " is not built"); + } + DataType keyType = getKeyType(columnName); + JavaRDD<String> secondaryKeys = + keys.select(col(columnName).cast(keyType).as(SECONDARY_KEY_COLUMN)) + .where(col(SECONDARY_KEY_COLUMN).isNotNull()) + .selectExpr(toSecondaryKey(keyType, SECONDARY_KEY_COLUMN)) + .as(Encoders.STRING()) + .toJavaRDD(); + // The reader only needs the metadata table enabled, and the table config above shows that the + // index exists. Without it, the factory returns a file system backed reader without indexes. + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder().enable(true).build(); + try (HoodieBackedTableMetadata tableMetadata = + (HoodieBackedTableMetadata) + metaClient + .getTableFormat() + .getMetadataFactory() + .create(engineContext, metaClient.getStorage(), 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 = + toJavaPairRDD(recordKeysBySecondaryKey) + .map( + recordKeyBySecondaryKey -> + toLookupResult( + dataPath, recordKeyBySecondaryKey._1(), recordKeyBySecondaryKey._2())); + StructType resultSchema = + new StructType() + .add(SECONDARY_KEY_COLUMN, DataTypes.StringType, false) + .add(FILE_COLUMN, DataTypes.StringType, false) + .add(POSITION_COLUMN, DataTypes.LongType, false); + return sparkSession + .createDataFrame(lookupResults, resultSchema) + .selectExpr( + fromSecondaryKey(keyType, SECONDARY_KEY_COLUMN) + + " AS `" + + columnName.replace("`", "``") + + "`", + FILE_COLUMN, + POSITION_COLUMN); + } + } + + /** + * Hudi returns an RDD backed result for keys in an RDD, but a list backed result when there are + * no keys to look up, for example when every key is null, and when the index has one file group. + * For one file group Hudi collects the keys and reads the file group on the driver, so the + * matches are collected there as well. + */ + private JavaPairRDD<String, String> toJavaPairRDD(HoodiePairData<String, String> pairData) { + if (pairData instanceof HoodieJavaPairRDD) { + return HoodieJavaPairRDD.getJavaPairRDD(pairData); + } + List<Tuple2<String, String>> pairs = + pairData.collectAsList().stream() + .map(pair -> new Tuple2<>(pair.getKey(), pair.getValue())) + .collect(Collectors.toList()); + return javaSparkContext.parallelizePairs(pairs); + } + + private static Row toLookupResult(String dataPath, String secondaryKey, String recordKey) { + // record keys generated by Hudi for files without record keys are "<relative path>_<row + // position>" + int positionSeparator = recordKey.lastIndexOf('_'); + if (positionSeparator == -1) { + throw new IllegalArgumentException( + "Expected a record key of the form <file path>_<row position> but got " + recordKey); + } + String relativeFilePath = recordKey.substring(0, positionSeparator); + return RowFactory.create( + secondaryKey, + FSUtils.constructAbsolutePath(dataPath, relativeFilePath).toString(), Review Comment: Checked with a Delta table whose partition values have a space and a `%`. The sync fails before the index is built, because the Delta source does not decode the encoded paths in the log, so the files Hudi reads do not exist. This is not specific to the index and is filed as #970. 1ab501dd adds a join of the lookup result to the table on `input_file_name()` to the Delta IT, and the special partition values can be added once #970 is fixed. ########## xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java: ########## @@ -745,9 +787,44 @@ private HoodieWriteConfig getWriteConfig( Schema schema, int numCommitsToKeep, int maxNumDeltaCommitsBeforeCompaction, - int timelineRetentionInHours) { + int timelineRetentionInHours, + Optional<String> secondaryIndexColumn) { Properties properties = new Properties(); properties.setProperty(HoodieMetadataConfig.AUTO_INITIALIZE.key(), "false"); + HoodieMetadataConfig.Builder metadataConfigBuilder = + HoodieMetadataConfig.newBuilder() + .enable(true) + .withProperties(properties) + // Build the column-stats index for all tables, but not the partition-stats + // index. Its generation path rebuilds a file-system view over the committed + // external parquet files and groups them by fileId, but XTable's + // externally-registered files have non-Hudi names whose fileId cannot be parsed + // once the "_hudiext" marker is stripped, which leads to failures on partitioned + // tables. + .withMetadataIndexColumnStats(true) + .withMetadataIndexPartitionStats(false) + .withMaxNumDeltaCommitsBeforeCompaction(maxNumDeltaCommitsBeforeCompaction); + // The secondary index maps a column value to the record key, so it needs the global record + // index to resolve the record key to a file. Rows without a record key are keyed by + // "<file path relative to the table>_<row position>", see + // https://github.com/apache/hudi/pull/19869. + if (targetConfig.getSecondaryIndexColumn().isPresent()) { Review Comment: Done in 1ab501dd. `secondaryIndexColumn` is removed and `getWriteConfig` reads it from `targetConfig`; `TestHudiConversionTarget` now asserts the built `HoodieWriteConfig` with and without the secondary index. ########## xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java: ########## @@ -35,22 +39,137 @@ 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"; + + /** Engines that {@link #EXECUTION_ENGINE} selects. */ + public enum ExecutionEngine { + JAVA, + SPARK; + + /** Returns the value of {@link #EXECUTION_ENGINE} that selects this engine. */ + public String getConfigValue() { + return name().toLowerCase(Locale.ROOT); + } + + private static ExecutionEngine fromConfigValue(String value) { + for (ExecutionEngine engine : values()) { + if (engine.getConfigValue().equalsIgnoreCase(value)) { + return engine; + } + } + throw new IllegalArgumentException( + String.format( + "Unsupported Hudi execution engine %s. Only %s are supported via %s.", + value, + Arrays.stream(values()) + .map(ExecutionEngine::getConfigValue) + .collect(Collectors.joining(" and ")), + EXECUTION_ENGINE)); + } + } + + /** + * Column to build a Hudi secondary index on. Setting it also enables the global record index, + * which the secondary index depends on. Both indexes are stored in the Hudi metadata table under + * the target table path and require table version 9. Only one column is supported, and the index + * is built when the Hudi metadata table is created; changing the column of an existing table is + * not supported yet. + */ + public static final String SECONDARY_INDEX_COLUMN = "xtable.hudi.target.secondary.index.column"; + + /** + * 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; + ExecutionEngine executionEngine; + Optional<String> secondaryIndexColumn; + 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); + ExecutionEngine executionEngine = + ExecutionEngine.fromConfigValue( + targetProperties + .getProperty(EXECUTION_ENGINE, ExecutionEngine.JAVA.getConfigValue()) + .trim()); + Optional<String> secondaryIndexColumn = + Optional.ofNullable(targetProperties.getProperty(SECONDARY_INDEX_COLUMN)) + .map(String::trim) + .filter(column -> !column.isEmpty()); + if (secondaryIndexColumn.isPresent() && tableVersion != HoodieTableVersion.NINE) { + throw new IllegalArgumentException( + String.format( + "%s requires Hudi target table version 9, set %s=9.", + SECONDARY_INDEX_COLUMN, HUDI_TABLE_VERSION)); + } + if (secondaryIndexColumn.filter(column -> column.contains(",")).isPresent()) { + throw new IllegalArgumentException( + String.format("%s supports only one column.", SECONDARY_INDEX_COLUMN)); + } + 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, + executionEngine, + secondaryIndexColumn, + 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 Optional<Integer> parsePositiveInt(Properties properties, String key) { + String configured = properties.getProperty(key); + if (configured == null || configured.trim().isEmpty()) { + return Optional.empty(); + } + int value = Integer.parseInt(configured.trim()); Review Comment: Done in 1ab501dd. A non-numeric value now fails with the key name, and a min file group count above the max is rejected, with tests in `TestHudiTargetConfig`. ########## xtable-core/src/main/java/org/apache/xtable/hudi/engine/HudiExecutionEngineProviderFactory.java: ########## @@ -0,0 +1,56 @@ +/* + * 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.hudi.engine; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +import org.apache.hadoop.conf.Configuration; +import org.apache.spark.serializer.KryoSerializer; +import org.apache.spark.sql.SparkSession; + +import org.apache.xtable.hudi.HudiTargetConfig; + +/** Creates the {@link HudiExecutionEngineProvider} selected by the Hudi target configuration. */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public class HudiExecutionEngineProviderFactory { + + public static HudiExecutionEngineProvider createProvider( + HudiTargetConfig targetConfig, Configuration configuration) { + if (targetConfig.getExecutionEngine() == HudiTargetConfig.ExecutionEngine.SPARK) { + return new SparkExecutionEngineProvider(getOrCreateSparkSession(configuration)); + } + return new JavaExecutionEngineProvider(configuration); + } + + /** + * Reuses the active Spark session when the sync runs inside a Spark application, otherwise starts + * a session that carries the Hadoop configuration of the sync. + */ + private static SparkSession getOrCreateSparkSession(Configuration configuration) { Review Comment: Done in 1ab501dd. Keys that start with `spark` are passed through as they are, the same as `DeltaConversionUtils.buildSparkSession`. -- 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]
