vinishjail97 commented on code in PR #935:
URL: https://github.com/apache/incubator-xtable/pull/935#discussion_r4252248831
##########
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:
`TestHudiConversionTarget` only uses the default config, so the write config
with a secondary index (global record index, `withWritesFileIdEncoding(1)`,
file group bounds, parallelism) is covered only by the IT. Would a unit test
that asserts the built `HoodieWriteConfig` be worth adding? Small nit: the
`secondaryIndexColumn` parameter is the same value as
`targetConfig.getSecondaryIndexColumn()`, so one of the two could go.
##########
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:
Nits: a non-numeric value surfaces as a bare `NumberFormatException` without
the key name, and a min file group count above the max is not rejected.
##########
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:
The record key is built from the decoded relative path, and
`constructAbsolutePath` rebuilds it here. I am not sure this round-trips for a
partition value with a space or `%`, which Delta stores URL-encoded in its log.
Could the Delta IT add one such partition value so a join on
`input_file_name()` is checked?
##########
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));
Review Comment:
The `xtable-utilities` bundle shades `hudi-java-client` but not
`hudi-spark-client`, so I think `execution.engine=spark` fails with
`NoClassDefFoundError` in `RunSync`. The default Java path should still load,
since the Spark classes resolve lazily, but every `xtable-core` test has Spark
on the classpath. Could a quick `RunSync` run with the bundled jar confirm the
Java path, and could the config Javadoc say that `spark` needs Hudi's Spark
bundle?
--
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]
