vinishjail97 commented on code in PR #935: URL: https://github.com/apache/incubator-xtable/pull/935#discussion_r4309846595
########## xtable-core/src/main/java/org/apache/xtable/hudi/engine/SparkExecutionEngineProvider.java: ########## @@ -0,0 +1,81 @@ +/* + * 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 java.util.List; +import java.util.Map; + +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SparkSession; + +import org.apache.hudi.client.SparkRDDWriteClient; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.table.HoodieSparkTable; +import org.apache.hudi.table.HoodieTable; + +/** + * Runs the Hudi commit and table services on Spark so the metadata table index generation is + * distributed across the cluster. + */ +public class SparkExecutionEngineProvider implements HudiExecutionEngineProvider { + private final HoodieSparkEngineContext engineContext; + + public SparkExecutionEngineProvider(SparkSession sparkSession) { + this.engineContext = + new HoodieSparkEngineContext( + JavaSparkContext.fromSparkContext(sparkSession.sparkContext())); + } + + @Override + public HoodieEngineContext getEngineContext() { + return engineContext; + } + + @Override + public HoodieTable<?, ?, ?, ?> createTable( + HoodieWriteConfig writeConfig, HoodieTableMetaClient metaClient) { + return HoodieSparkTable.create(writeConfig, engineContext, metaClient); + } + + @Override + public void commit( + HoodieWriteConfig writeConfig, + String instantTime, + List<WriteStatus> writeStatuses, + Option<Map<String, String>> extraMetadata, + String commitActionType, + Map<String, List<String>> partitionToReplacedFileIds) { + try (SparkRDDWriteClient<?> writeClient = + new SparkRDDWriteClient<>(engineContext, writeConfig)) { + writeClient.setOperationType(WriteOperationType.UNKNOWN); Review Comment: `UNKNOWN` is Hudi's default for commit metadata ([HoodieCommitMetadata.java#L78](https://github.com/apache/hudi/blob/a9185b7c4e83422132c25f2efdf282c5bc6af2c7/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieCommitMetadata.java#L78)), and #772 already set it in `HudiConversionTarget`. An XTable commit both adds and replaces files, so `INSERT`/`UPSERT` would be inaccurate, and the overwrite types are rejected by the secondary index ([L1740-L1741](https://github.com/apache/hudi/blob/a9185b7c4e83422132c25f2efdf282c5bc6af2c7/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java#L1740-L1741)) and trigger the replaced-records path of the record index ([L2453](https://github.com/apache/hudi/blob/a9185b7c4e83422132c25f2efdf282c5bc6af2c7/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java#L2453)). ########## xtable-core/src/main/java/org/apache/xtable/hudi/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')"; Review Comment: It has to stay LTZ: `timestamp_micros` returns `TIMESTAMP` ([datetimeExpressions.scala#L568](https://github.com/apache/spark/blob/7c14a3c28b141cc97a330c4d0f5d2a6da7267f85/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala#L568)), and a mixed subtraction casts the LTZ side to NTZ in the session time zone ([TypeCoercion.scala#L1183-L1187](https://github.com/apache/spark/blob/7c14a3c28b141cc97a330c4d0f5d2a6da7267f85/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala#L1183-L1187)). With an NTZ epoch, `2024-01-02 03:04:05.123456` comes back as `2024-01-01 19:04:05.123456` in `America/Los_Angeles`, while the LTZ difference does not depend on the time zone. ########## xtable-core/src/main/java/org/apache/xtable/hudi/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); Review Comment: It is reachable with the Spark engine context too. Hudi returns list-backed data for an empty key set ([L912-L915](https://github.com/apache/hudi/blob/a9185b7c4e83422132c25f2efdf282c5bc6af2c7/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java#L912-L915)) and for an index with one file group ([L295-L296](https://github.com/apache/hudi/blob/a9185b7c4e83422132c25f2efdf282c5bc6af2c7/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java#L295-L296)), and `getJavaPairRDD` throws `ClassCastException` on both; the IT covers them with empty and null-only keys. ########## xtable-core/src/main/java/org/apache/xtable/index/HudiBackedSecondaryIndexUtils.java: ########## @@ -0,0 +1,139 @@ +/* + * 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; Review Comment: Done in 847126c3, the Hudi backed classes moved to `org.apache.xtable.hudi.index`, and `IndexReader` stays in `org.apache.xtable.index`. ########## xtable-core/src/main/java/org/apache/xtable/index/IndexReader.java: ########## @@ -0,0 +1,48 @@ +/* + * 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; + +/** + * Reads an index that an {@link IndexWriter} maintains, so an engine can find the rows that hold + * given values of a column without a full table scan or join. The key and result types depend on + * the engine that runs the lookup, for example a Spark {@code Dataset<Row>}. + * + * @param <T> The type that identifies the indexed table + * @param <K> The type of the keys to look up + * @param <R> The type of the lookup result + */ +public interface IndexReader<T, K, R> { Review Comment: Done in 847126c3, `IndexReader` is marked `@Evolving`. ########## xtable-core/src/main/java/org/apache/xtable/index/HudiBackedSecondaryIndexWriter.java: ########## @@ -0,0 +1,181 @@ +/* + * 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.util.Collections; +import java.util.Properties; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; + +import org.apache.hudi.common.table.HoodieTableVersion; + +import com.google.common.base.Preconditions; + +import org.apache.xtable.conversion.ConversionConfig; +import org.apache.xtable.conversion.ConversionController; +import org.apache.xtable.conversion.ConversionSourceProvider; +import org.apache.xtable.conversion.SourceTable; +import org.apache.xtable.conversion.TargetTable; +import org.apache.xtable.hudi.HudiTargetConfig; +import org.apache.xtable.model.storage.TableFormat; +import org.apache.xtable.model.sync.SyncResult; +import org.apache.xtable.model.sync.SyncStatusCode; + +/** + * Builds and maintains a secondary index for a table of any source format that XTable reads, backed + * by the Hudi metadata table. {@link HudiBackedSecondaryIndexReader} reads the index. + * + * <p>The index is built by syncing the table to Hudi with the record index and a secondary index + * enabled; the Hudi metadata lives under {@code <data path>/.hoodie/metadata}, where the data path + * is {@link SourceTable#getDataPath()}. Because the data files carry no Hudi record keys, the + * record key of a row is {@code <file path relative to the data path>_<row position>}. Hudi + * requires every data file to be under the data path, so files outside it are not supported. + * + * <p>The parts that depend on the source format are parameters: the {@link + * ConversionSourceProvider} reads the table, its schema gives the type of the indexed column, and + * the caller sets the data path of the {@link SourceTable}, for example with {@link + * org.apache.xtable.catalog.TableFormatUtils#getTableDataLocation} for an Iceberg table. Hudi + * tables are not supported as a source, because they have their own metadata table. + * + * <p>The sync runs Hudi's commit on its Spark engine, which reuses the active Spark session or + * starts one, because Hudi's Java engine cannot build the secondary index yet. The API of this + * class does not depend on Spark. + */ +@Log4j2 +public class HudiBackedSecondaryIndexWriter implements IndexWriter<SourceTable> { + private final String tableLocation; + private final String dataPath; + private final SourceTable sourceTable; + private final ConversionSourceProvider<?> sourceProvider; + private final Configuration configuration; + private final Properties targetTableProperties; + private final String indexedColumn; + + /** + * @param sourceTable the table to index. A table of a catalog sets its catalog config, name and + * namespace. The Hudi table is created under its data path + * @param sourceProvider the provider of the conversion source for the format of the table. The + * index initializes it with the given configuration + * @param configuration the Hadoop configuration used to read the table and write the index + * @param targetTableProperties {@link HudiTargetConfig} properties for the sync. {@link + * HudiTargetConfig#SECONDARY_INDEX_COLUMN} names the column to index, and other properties + * such as the record index file group counts are optional + */ + public HudiBackedSecondaryIndexWriter( + SourceTable sourceTable, + ConversionSourceProvider<?> sourceProvider, + Configuration configuration, + Properties targetTableProperties) { + Preconditions.checkArgument( + !TableFormat.HUDI.equals(sourceTable.getFormatName()), + "A Hudi table has its own metadata table and cannot be the source of the index"); + this.sourceTable = sourceTable; + this.tableLocation = sourceTable.getBasePath(); + this.dataPath = sourceTable.getDataPath(); + this.configuration = configuration; + this.targetTableProperties = getIndexTargetProperties(targetTableProperties); + this.indexedColumn = + HudiTargetConfig.fromProperties(this.targetTableProperties) + .getSecondaryIndexColumn() + .orElseThrow( + () -> + new IllegalArgumentException( + "Set " + + HudiTargetConfig.SECONDARY_INDEX_COLUMN + + " to the column to index")); + this.sourceProvider = sourceProvider; + sourceProvider.init(configuration); + HudiBackedSecondaryIndexUtils.getIndexableField( + HudiBackedSecondaryIndexUtils.readSchema(sourceProvider, sourceTable), indexedColumn); + } + + /** + * Returns the given {@link HudiTargetConfig} properties with the settings the index requires: the + * Spark engine and table version 9. + */ + private static Properties getIndexTargetProperties(Properties targetTableProperties) { + Properties indexTargetProperties = new Properties(); + indexTargetProperties.putAll(targetTableProperties); + indexTargetProperties.setProperty( + HudiTargetConfig.EXECUTION_ENGINE, HudiTargetConfig.ExecutionEngine.SPARK.getConfigValue()); + // the secondary index is only available from Hudi table version 8 onwards + indexTargetProperties.setProperty( + HudiTargetConfig.HUDI_TABLE_VERSION, String.valueOf(HoodieTableVersion.NINE.versionCode())); + return indexTargetProperties; + } + + /** + * Syncs the table to Hudi up to the latest commit at the start of the sync. The sync applies the + * new commits incrementally, or syncs a full snapshot when an incremental sync is not safe. It + * writes the record index and the secondary index of the configured column together with the Hudi + * table. A sync with no new commit changes nothing. + * + * @throws IllegalStateException when the sync fails, or when the index does not exist after it, + * for example because the indexed column changed + */ + @Override + public void syncIndex(SourceTable table) { + HudiBackedSecondaryIndexUtils.checkIndexedTable(tableLocation, table); + // The sync reports a failed target in its result instead of throwing. It returns no result + // for the target when the table has no new commits. + SyncResult syncResult = Review Comment: Agreed, there is no need for one. 847126c3 removes `IndexWriter` and the writer, so the index is built by a regular Hudi target sync with `xtable.hudi.target.secondary.index.column`, and `RunSync` now passes per-table `targetProperties`. The reader exposes `getLastSyncedSourceIdentifier()` from the last commit's `XTABLE_METADATA`, so a caller can check that the index is at the source snapshot it reads. ########## xtable-core/src/test/java/org/apache/xtable/index/TestMetastoreCatalog.java: ########## @@ -0,0 +1,146 @@ +/* + * 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.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +import org.apache.hadoop.conf.Configuration; + +import org.apache.iceberg.BaseMetastoreCatalog; +import org.apache.iceberg.BaseMetastoreTableOperations; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.hadoop.HadoopFileIO; +import org.apache.iceberg.io.FileIO; + +/** + * A catalog that keeps the current metadata location of each table in memory, like Glue or a Hive + * Metastore keeps it in its own store. Its metadata files have UUID based names and there is no + * version hint, so {@code HadoopTables} cannot load its tables from their location. The state is + * static, because XTable creates its own catalog instance from the class name. + */ +public class TestMetastoreCatalog extends BaseMetastoreCatalog { + private static final Map<String, String> METADATA_LOCATIONS = new ConcurrentHashMap<>(); + + private String catalogName; + private String warehouseLocation; + private FileIO fileIO; + + @Override + public void initialize(String name, Map<String, String> properties) { + this.catalogName = name; + this.warehouseLocation = properties.get(CatalogProperties.WAREHOUSE_LOCATION); + this.fileIO = new HadoopFileIO(new Configuration()); + } + + @Override + public String name() { + return catalogName; + } + + @Override + protected TableOperations newTableOps(TableIdentifier tableIdentifier) { + return new TestMetastoreTableOperations(key(tableIdentifier), fileIO); + } + + @Override + protected String defaultWarehouseLocation(TableIdentifier tableIdentifier) { + return warehouseLocation + + "/" + + String.join("/", tableIdentifier.namespace().levels()) + + "/" + + tableIdentifier.name(); + } + + @Override + public List<TableIdentifier> listTables(Namespace namespace) { + String prefix = catalogName + "." + namespace + "."; + return METADATA_LOCATIONS.keySet().stream() + .filter(key -> key.startsWith(prefix)) + .map(key -> TableIdentifier.of(namespace, key.substring(prefix.length()))) + .collect(Collectors.toList()); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + return METADATA_LOCATIONS.remove(key(identifier)) != null; + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + throw new UnsupportedOperationException("Renaming tables is not supported"); + } + + private String key(TableIdentifier tableIdentifier) { + return catalogName + "." + tableIdentifier; + } + + private static class TestMetastoreTableOperations extends BaseMetastoreTableOperations { + private final String key; + private final FileIO fileIO; + + TestMetastoreTableOperations(String key, FileIO fileIO) { + this.key = key; + this.fileIO = fileIO; + } + + @Override + protected String tableName() { + return key; + } + + @Override + public FileIO io() { + return fileIO; + } + + @Override + protected void doRefresh() { + String metadataLocation = METADATA_LOCATIONS.get(key); + if (metadataLocation == null) { + disableRefresh(); + } else { + refreshFromMetadataLocation(metadataLocation); + } + } + + @Override + protected void doCommit(TableMetadata base, TableMetadata metadata) { + String newMetadataLocation = writeNewMetadataIfRequired(base == null, metadata); + String expectedLocation = base == null ? null : base.metadataFileLocation(); + METADATA_LOCATIONS.compute( + key, + (ignored, currentLocation) -> { + if (!java.util.Objects.equals(currentLocation, expectedLocation)) { Review Comment: Done in 847126c3. -- 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]
