> hudiTableSchemaOpt)
+ {
+ this.tableHandle = requireNonNull(tableHandle, "table is null");
this.dataColumns = requireNonNull(dataColumns, "dataColumns is null");
this.partitionColumns = requireNonNull(partitionColumns, "partitionColumns is null");
this.tableParameters = requireNonNull(tableParameters, "tableParameters is null");
- this.tupleDomain = requireNonNull(tupleDomain, "tupleDomain is null");
+ this.regularPredicates = requireNonNull(regularPredicates, "regularPredicates is null");
+ this.partitionPredicates = requireNonNull(partitionPredicates, "partitionPredicates is null");
+ this.hudiTableSchemaOpt = requireNonNull(hudiTableSchemaOpt, "hudiTableSchemaOpt is null");
+ }
+
+ /**
+ * Builds a lazily-parsed Avro schema from the given schema string.
+ *
+ * Returns {@code Optional.empty()} if the input string is null/empty
+ * or if parsing the schema fails.
+ */
+ private static Optional> buildTableSchema(String tableSchemaStr)
+ {
+ if (StringUtils.isNullOrEmpty(tableSchemaStr)) {
+ return Optional.empty();
+ }
+
+ try {
+ Lazy lazySchema = Lazy.lazily(() -> new Schema.Parser().parse(tableSchemaStr));
+ return Optional.of(lazySchema);
+ }
+ catch (Exception e) {
+ log.warn(e, "Failed to parse table schema: %s", tableSchemaStr);
+ return Optional.empty();
+ }
}
@JsonProperty
- public HudiTableHandle getTable()
+ public HudiTableHandle getTableHandle()
{
- return table;
+ return tableHandle;
}
@JsonProperty
@@ -75,9 +121,30 @@ public Map getTableParameters()
}
@JsonProperty
- public TupleDomain getTupleDomain()
+ public TupleDomain getRegularPredicates()
+ {
+ return regularPredicates;
+ }
+
+ @JsonProperty
+ public TupleDomain getPartitionPredicates()
+ {
+ return partitionPredicates;
+ }
+
+ @JsonProperty
+ public String getTableSchemaStr()
+ {
+ return hudiTableSchemaOpt
+ .map(Lazy::get)
+ .map(Schema::toString)
+ .orElse("");
+ }
+
+ @JsonIgnore
+ public Schema getTableSchema()
{
- return tupleDomain;
+ return hudiTableSchemaOpt.map(Lazy::get).orElse(null);
}
@Override
@@ -90,19 +157,20 @@ public boolean equals(Object o)
return false;
}
HudiTableLayoutHandle that = (HudiTableLayoutHandle) o;
- return Objects.equals(table, that.table) &&
- Objects.equals(tupleDomain, that.tupleDomain);
+ return Objects.equals(tableHandle, that.tableHandle) &&
+ Objects.equals(regularPredicates, that.regularPredicates) &&
+ Objects.equals(partitionPredicates, that.partitionPredicates);
}
@Override
public int hashCode()
{
- return Objects.hash(table, tupleDomain);
+ return Objects.hash(tableHandle, regularPredicates, partitionPredicates);
}
@Override
public String toString()
{
- return table.toString();
+ return tableHandle.toString();
}
}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/HiveStylePartitionValueExtractor.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/HiveStylePartitionValueExtractor.java
new file mode 100644
index 0000000000000..cffa3cce37239
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/HiveStylePartitionValueExtractor.java
@@ -0,0 +1,24 @@
+package com.facebook.presto.hudi.partition;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Extractor for Hive Style Partitioned tables, when the partition folders are key value pairs.
+ *
+ * This implementation extracts the partition value of yyyy-mm-dd from the path of type datestr=yyyy-mm-dd.
+ */
+public class HiveStylePartitionValueExtractor implements PartitionValueExtractor {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public List extractPartitionValuesInPath(String partitionPath) {
+ // partition path is expected to be in this format partition_key=partition_value.
+ String[] splits = partitionPath.split("=");
+ if (splits.length != 2) {
+ throw new IllegalArgumentException(
+ "Partition path " + partitionPath + " is not in the form partition_key=partition_value.");
+ }
+ return Collections.singletonList(splits[1]);
+ }
+}
\ No newline at end of file
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/MultiPartKeysValueExtractor.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/MultiPartKeysValueExtractor.java
new file mode 100644
index 0000000000000..9058fe4fd9f8f
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/MultiPartKeysValueExtractor.java
@@ -0,0 +1,32 @@
+package com.facebook.presto.hudi.partition;
+
+import org.apache.hudi.common.util.ValidationUtils;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Partition Key extractor treating each value delimited by slash as separate key.
+ */
+public class MultiPartKeysValueExtractor implements PartitionValueExtractor {
+
+ @Override
+ public List extractPartitionValuesInPath(String partitionPath) {
+ // If the partitionPath is empty string( which means none-partition table), the partition values
+ // should be empty list.
+ if (partitionPath.isEmpty()) {
+ return Collections.emptyList();
+ }
+ String[] splits = partitionPath.split("/");
+ return Arrays.stream(splits).map(s -> {
+ if (s.contains("=")) {
+ String[] moreSplit = s.split("=");
+ ValidationUtils.checkArgument(moreSplit.length == 2, "Partition Field (" + s + ") not in expected format");
+ return moreSplit[1];
+ }
+ return s;
+ }).collect(Collectors.toList());
+ }
+}
\ No newline at end of file
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/NonPartitionedExtractor.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/NonPartitionedExtractor.java
new file mode 100644
index 0000000000000..e4ddbd656eea3
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/NonPartitionedExtractor.java
@@ -0,0 +1,15 @@
+package com.facebook.presto.hudi.partition;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Extractor for Non-partitioned hive tables.
+ */
+public class NonPartitionedExtractor implements PartitionValueExtractor {
+
+ @Override
+ public List extractPartitionValuesInPath(String partitionPath) {
+ return new ArrayList<>();
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/PartitionValueExtractor.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/PartitionValueExtractor.java
new file mode 100644
index 0000000000000..fc0e0338a5da1
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/PartitionValueExtractor.java
@@ -0,0 +1,16 @@
+package com.facebook.presto.hudi.partition;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * HDFS Path contain hive partition values for the keys it is partitioned on. This mapping is not straight forward and
+ * requires a pluggable implementation to extract the partition value from HDFS path.
+ *
+ * e.g. Hive table partitioned by datestr=yyyy-mm-dd and hdfs path /app/hoodie/dataset1/YYYY=[yyyy]/MM=[mm]/DD=[dd]
+ */
+public interface PartitionValueExtractor extends Serializable
+{
+
+ List extractPartitionValuesInPath(String partitionPath);
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/SinglePartPartitionValueExtractor.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/SinglePartPartitionValueExtractor.java
new file mode 100644
index 0000000000000..ced4b6e387575
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/partition/SinglePartPartitionValueExtractor.java
@@ -0,0 +1,19 @@
+package com.facebook.presto.hudi.partition;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Extractor for a partition path from a single column.
+ *
+ * This implementation extracts the partition value from the partition path as a single part
+ * even if the relative partition path contains slashes, e.g., the `TimestampBasedKeyGenerator`
+ * transforms the timestamp column into the partition path in the format of "yyyyMM/dd/HH".
+ * The slash (`/`) is replaced with dash (`-`), e.g., `202210/01/20` -> `202210-01-20`.
+ */
+public class SinglePartPartitionValueExtractor implements PartitionValueExtractor {
+ @Override
+ public List extractPartitionValuesInPath(String partitionPath) {
+ return Collections.singletonList(partitionPath.replace('/', '-'));
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/HudiDirectoryLister.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/HudiDirectoryLister.java
new file mode 100644
index 0000000000000..3ffa424eee7d6
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/HudiDirectoryLister.java
@@ -0,0 +1,13 @@
+package com.facebook.presto.hudi.query;
+
+import com.facebook.presto.hudi.HudiPartition;
+import org.apache.hudi.common.model.FileSlice;
+
+import java.io.Closeable;
+import java.util.stream.Stream;
+
+public interface HudiDirectoryLister
+ extends Closeable
+{
+ Stream listStatus(HudiPartition hudiPartition, boolean useIndex);
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/HudiSnapshotDirectoryLister.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/HudiSnapshotDirectoryLister.java
new file mode 100644
index 0000000000000..362811857f92e
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/HudiSnapshotDirectoryLister.java
@@ -0,0 +1,88 @@
+package com.facebook.presto.hudi.query;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.presto.hudi.HudiPartition;
+import com.facebook.presto.hudi.HudiTableHandle;
+import com.facebook.presto.hudi.HudiTableLayoutHandle;
+import com.facebook.presto.hudi.query.index.HudiIndexSupport;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.SchemaTableName;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.util.Lazy;
+
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import static com.facebook.presto.hudi.HudiSessionProperties.isHudiMetadataTableEnabled;
+import static com.facebook.presto.hudi.query.index.IndexSupportFactory.createIndexSupport;
+import static org.apache.hudi.common.table.view.FileSystemViewManager.createInMemoryFileSystemViewWithTimeline;
+
+public class HudiSnapshotDirectoryLister
+ implements HudiDirectoryLister
+{
+ private static final Logger log = Logger.get(HudiSnapshotDirectoryLister.class);
+ private final HudiTableHandle tableHandle;
+ private final Lazy lazyFileSystemView;
+ private final Optional indexSupportOpt;
+
+ public HudiSnapshotDirectoryLister(
+ ConnectorSession session,
+ HudiTableLayoutHandle layoutHandle,
+ boolean enableMetadataTable,
+ Lazy lazyTableMetadata)
+ {
+ this.tableHandle = layoutHandle.getTableHandle();
+ SchemaTableName schemaTableName = tableHandle.getSchemaTableName();
+ this.lazyFileSystemView = Lazy.lazily(() -> {
+ HoodieTimer timer = HoodieTimer.start();
+ HoodieTableMetaClient metaClient = layoutHandle.getTableHandle().getMetaClient();
+ HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder().enable(isHudiMetadataTableEnabled(session)).build();
+ HoodieTimeline timeline = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ HoodieLocalEngineContext engineContext = new HoodieLocalEngineContext(metaClient.getStorageConf());
+ HoodieTableFileSystemView fsView = createInMemoryFileSystemViewWithTimeline(engineContext, metaClient, metadataConfig, timeline);
+ log.info("Created file system view of table %s in %s ms", schemaTableName, timer.endTimer());
+ return fsView;
+ });
+
+ Lazy lazyMetaClient = Lazy.lazily(tableHandle::getMetaClient);
+ this.indexSupportOpt = enableMetadataTable ?
+ createIndexSupport(layoutHandle, lazyMetaClient, lazyTableMetadata, layoutHandle.getRegularPredicates(), session) : Optional.empty();
+ }
+
+ @Override
+ public Stream listStatus(HudiPartition hudiPartition, boolean useIndex)
+ {
+ HoodieTimer timer = HoodieTimer.start();
+ Stream slices = lazyFileSystemView.get().getLatestFileSlicesBeforeOrOn(
+ hudiPartition.getRelativePartitionPath(),
+ tableHandle.getLatestCommitTime(),
+ false);
+
+ if (!useIndex) {
+ return slices;
+ }
+
+ Stream fileSlices = slices
+ .filter(slice -> indexSupportOpt
+ .map(indexSupport -> !indexSupport.shouldSkipFileSlice(slice))
+ .orElse(true));
+ log.info("Listed partition [%s] on table %s.%s in %s ms",
+ hudiPartition, tableHandle.getSchemaName(), tableHandle.getTableName(), timer.endTimer());
+ return fileSlices;
+ }
+
+ @Override
+ public void close()
+ {
+ if (!lazyFileSystemView.get().isClosed()) {
+ lazyFileSystemView.get().close();
+ }
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiBaseIndexSupport.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiBaseIndexSupport.java
new file mode 100644
index 0000000000000..3c2738996fc91
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiBaseIndexSupport.java
@@ -0,0 +1,53 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.presto.spi.SchemaTableName;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieIndexDefinition;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.util.Lazy;
+
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+public abstract class HudiBaseIndexSupport
+ implements HudiIndexSupport
+{
+ private final Logger log;
+ protected final SchemaTableName schemaTableName;
+ protected final Lazy lazyMetaClient;
+
+ public HudiBaseIndexSupport(Logger log, SchemaTableName schemaTableName, Lazy lazyMetaClient)
+ {
+ this.log = requireNonNull(log, "log is null");
+ this.schemaTableName = requireNonNull(schemaTableName, "schemaTableName is null");
+ this.lazyMetaClient = requireNonNull(lazyMetaClient, "metaClient is null");
+ }
+
+ public void printDebugMessage(Map> candidateFileSlices, Map> inputFileSlices, long lookupDurationMs)
+ {
+ if (log.isDebugEnabled()) {
+ int candidateFileSize = candidateFileSlices.values().stream().mapToInt(List::size).sum();
+ int totalFiles = inputFileSlices.values().stream().mapToInt(List::size).sum();
+ double skippingPercent = totalFiles == 0 ? 0.0d : (totalFiles - candidateFileSize) / (totalFiles * 1.0d);
+
+ log.info("Total files: %s; files after data skipping: %s; skipping percent %s; time taken: %s ms; table name: %s",
+ totalFiles,
+ candidateFileSize,
+ skippingPercent,
+ lookupDurationMs,
+ schemaTableName);
+ }
+ }
+
+ protected Map getAllIndexDefinitions()
+ {
+ if (lazyMetaClient.get().getIndexMetadata().isEmpty()) {
+ return Map.of();
+ }
+
+ return lazyMetaClient.get().getIndexMetadata().get().getIndexDefinitions();
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiColumnStatsIndexSupport.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiColumnStatsIndexSupport.java
new file mode 100644
index 0000000000000..b702772e20142
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiColumnStatsIndexSupport.java
@@ -0,0 +1,324 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.airlift.units.Duration;
+import com.facebook.presto.common.predicate.Domain;
+import com.facebook.presto.common.predicate.Range;
+import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.common.predicate.ValueSet;
+import com.facebook.presto.common.type.Type;
+import com.facebook.presto.common.type.VarcharType;
+import com.facebook.presto.hudi.util.TupleDomainUtils;
+import com.facebook.presto.parquet.predicate.TupleDomainParquetPredicate;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.SchemaTableName;
+import com.google.common.collect.ImmutableList;
+import io.airlift.slice.Slice;
+import io.airlift.slice.Slices;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.avro.model.HoodieMetadataColumnStats;
+import org.apache.hudi.common.model.BaseFile;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieIndexDefinition;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.hash.ColumnIndexID;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.util.Lazy;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+import static com.facebook.presto.common.type.BigintType.BIGINT;
+import static com.facebook.presto.common.type.BooleanType.BOOLEAN;
+import static com.facebook.presto.common.type.DateType.DATE;
+import static com.facebook.presto.common.type.DoubleType.DOUBLE;
+import static com.facebook.presto.common.type.IntegerType.INTEGER;
+import static com.facebook.presto.common.type.RealType.REAL;
+import static com.facebook.presto.common.type.SmallintType.SMALLINT;
+import static com.facebook.presto.common.type.TinyintType.TINYINT;
+import static com.facebook.presto.hudi.HudiSessionProperties.getColumnStatsWaitTimeout;
+import static com.facebook.presto.parquet.predicate.PredicateUtils.isStatisticsOverflow;
+import static java.lang.Float.floatToRawIntBits;
+
+public class HudiColumnStatsIndexSupport
+ extends HudiBaseIndexSupport
+{
+ private static final Logger log = Logger.get(HudiColumnStatsIndexSupport.class);
+ // file name -> column name -> domain with column stats
+ private final CompletableFuture>>> domainsWithStatsFuture;
+ protected final TupleDomain regularColumnPredicates;
+ private final List regularColumns;
+ private final Duration columnStatsWaitTimeout;
+ private final long futureStartTimeMs;
+
+ public HudiColumnStatsIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy lazyMetaClient, Lazy lazyTableMetadata, TupleDomain regularColumnPredicates)
+ {
+ this(log, session, schemaTableName, lazyMetaClient, lazyTableMetadata, regularColumnPredicates);
+ }
+
+ public HudiColumnStatsIndexSupport(Logger log, ConnectorSession session, SchemaTableName schemaTableName, Lazy lazyMetaClient, Lazy lazyTableMetadata, TupleDomain regularColumnPredicates)
+ {
+ super(log, schemaTableName, lazyMetaClient);
+ this.columnStatsWaitTimeout = getColumnStatsWaitTimeout(session);
+ this.regularColumnPredicates = regularColumnPredicates;
+ this.regularColumns = regularColumnPredicates.getDomains()
+ .map(domains -> new ArrayList<>(domains.keySet()))
+ .orElse(new ArrayList<>());
+ if (regularColumnPredicates.isAll() || regularColumnPredicates.getDomains().isEmpty()) {
+ this.domainsWithStatsFuture = CompletableFuture.completedFuture(Optional.empty());
+ }
+ else {
+ // Get filter columns
+ List encodedTargetColumnNames = regularColumns
+ .stream()
+ .map(col -> new ColumnIndexID(col).asBase64EncodedString()).collect(Collectors.toList());
+
+ Map columnTypes = regularColumnPredicates.getDomains().get().entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getType()));
+
+ domainsWithStatsFuture = CompletableFuture.supplyAsync(() -> {
+ HoodieTimer timer = HoodieTimer.start();
+ if (!lazyMetaClient.get().getTableConfig().getMetadataPartitions()
+ .contains(HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS)) {
+ return Optional.empty();
+ }
+
+ Map> domainsWithStats =
+ lazyTableMetadata.get().getRecordsByKeyPrefixes(encodedTargetColumnNames,
+ HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, true)
+ .collectAsList()
+ .stream()
+ .filter(f -> f.getData().getColumnStatMetadata().isPresent())
+ .map(f -> f.getData().getColumnStatMetadata().get())
+ .collect(Collectors.groupingBy(
+ HoodieMetadataColumnStats::getFileName,
+ Collectors.toMap(
+ HoodieMetadataColumnStats::getColumnName,
+ // Pre-compute the Domain object for each HoodieMetadataColumnStats
+ stats -> getDomainFromColumnStats(stats.getColumnName(), columnTypes.get(stats.getColumnName()), stats))));
+
+ log.debug("Column stats lookup took %s ms and identified %d relevant file IDs.", timer.endTimer(), domainsWithStats.size());
+
+ return Optional.of(domainsWithStats);
+ });
+ }
+ this.futureStartTimeMs = System.currentTimeMillis();
+ }
+
+ @Override
+ public boolean shouldSkipFileSlice(FileSlice slice)
+ {
+ try {
+ if (domainsWithStatsFuture.isDone()) {
+ Optional>> domainsWithStatsOpt = domainsWithStatsFuture.get();
+ return domainsWithStatsOpt
+ .map(domainsWithStats -> shouldSkipFileSlice(slice, domainsWithStats, regularColumnPredicates, regularColumns))
+ .orElse(false);
+ }
+
+ long elapsedMs = System.currentTimeMillis() - futureStartTimeMs;
+ if (elapsedMs > columnStatsWaitTimeout.toMillis()) {
+ // Took too long; skip decision
+ return false;
+ }
+
+ // If still within the timeout window, wait up to the remaining time
+ long remainingMs = Math.max(0, columnStatsWaitTimeout.toMillis() - elapsedMs);
+ Optional>> statsOpt =
+ domainsWithStatsFuture.get(remainingMs, TimeUnit.MILLISECONDS);
+
+ return statsOpt
+ .map(stats -> shouldSkipFileSlice(slice, stats, regularColumnPredicates, regularColumns))
+ .orElse(false);
+ }
+ catch (TimeoutException | InterruptedException | ExecutionException e) {
+ return false;
+ }
+ }
+
+ @Override
+ public boolean canApply(TupleDomain tupleDomain)
+ {
+ boolean isIndexSupported = isIndexSupportAvailable();
+ // indexDefinition is only available after table version EIGHT
+ // For tables that have versions < EIGHT, column stats index is available as long as partition in metadata is available
+ if (!isIndexSupported || lazyMetaClient.get().getTableConfig().getTableVersion().lesserThan(HoodieTableVersion.EIGHT)) {
+ log.debug("Column Stats Index partition is not enabled in metadata.");
+ return isIndexSupported;
+ }
+
+ Map indexDefinitions = getAllIndexDefinitions();
+ HoodieIndexDefinition colStatsDefinition = indexDefinitions.get(HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS);
+ if (colStatsDefinition == null || colStatsDefinition.getSourceFields() == null || colStatsDefinition.getSourceFields().isEmpty()) {
+ log.warn("Column stats index definition is missing or has no source fields defined");
+ return false;
+ }
+
+ // Optimization applied: Only consider applicable if predicates reference indexed columns
+ List sourceFields = colStatsDefinition.getSourceFields();
+ boolean applicable = TupleDomainUtils.areSomeFieldsReferenced(tupleDomain, sourceFields);
+
+ if (applicable) {
+ log.debug("Column Stats Index is available and applicable (predicates reference indexed columns).");
+ }
+ else {
+ log.debug("Column Stats Index is available, but predicates do not reference any indexed columns.");
+ }
+ return applicable;
+ }
+
+ public boolean isIndexSupportAvailable()
+ {
+ return lazyMetaClient.get().getTableConfig().getMetadataPartitions()
+ .contains(HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS);
+ }
+
+ // TODO: Move helper functions below to TupleDomain/DomainUtils
+ private static boolean shouldSkipFileSlice(
+ FileSlice fileSlice,
+ Map> domainsWithStats,
+ TupleDomain regularColumnPredicates,
+ List regularColumns)
+ {
+ List filesToLookUp = new ArrayList<>();
+ fileSlice.getBaseFile()
+ .map(BaseFile::getFileName)
+ .ifPresent(filesToLookUp::add);
+
+ if (fileSlice.hasLogFiles()) {
+ fileSlice.getLogFiles().forEach(logFile -> filesToLookUp.add(logFile.getFileName()));
+ }
+
+ // if any log or base file in the file slice matches the predicate, all files in the file slice needs to be read
+ return filesToLookUp.stream().allMatch(fileName -> {
+ // If no stats exist for this specific file, we cannot prune it.
+ if (!domainsWithStats.containsKey(fileName)) {
+ return false;
+ }
+ Map fileDomainsWithStats = domainsWithStats.get(fileName);
+ return !evaluateStatisticPredicate(regularColumnPredicates, fileDomainsWithStats, regularColumns);
+ });
+ }
+
+ protected static boolean evaluateStatisticPredicate(
+ TupleDomain regularColumnPredicates,
+ Map domainsWithStats,
+ List regularColumns)
+ {
+ if (regularColumnPredicates.isNone() || !regularColumnPredicates.getDomains().isPresent()) {
+ return true;
+ }
+ for (String regularColumn : regularColumns) {
+ Domain columnPredicate = regularColumnPredicates.getDomains().get().get(regularColumn);
+ Optional currentColumnStats = Optional.ofNullable(domainsWithStats.get(regularColumn));
+ if (currentColumnStats.isEmpty()) {
+ // No stats for column
+ }
+ else {
+ Domain domain = currentColumnStats.get();
+ if (columnPredicate.intersect(domain).isNone()) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ static Domain getDomainFromColumnStats(String colName, Type type, HoodieMetadataColumnStats statistics)
+ {
+ if (statistics == null) {
+ return Domain.all(type);
+ }
+ boolean hasNullValue = statistics.getNullCount() != 0L;
+ boolean hasNonNullValue = statistics.getValueCount() - statistics.getNullCount() > 0;
+ if (!hasNonNullValue || statistics.getMaxValue() == null || statistics.getMinValue() == null) {
+ return Domain.create(ValueSet.all(type), hasNullValue);
+ }
+ if (!(statistics.getMinValue() instanceof GenericRecord) ||
+ !(statistics.getMaxValue() instanceof GenericRecord)) {
+ return Domain.all(type);
+ }
+ return getDomainFromColumnStats(colName, type, ((GenericRecord) statistics.getMinValue()).get(0),
+ ((GenericRecord) statistics.getMaxValue()).get(0), hasNullValue);
+ }
+
+ /**
+ * Get a domain for the ranges defined by each pair of elements from {@code minimums} and {@code maximums}.
+ * Both arrays must have the same length.
+ */
+ private static Domain getDomainFromColumnStats(String colName, Type type, Object minimum, Object maximum, boolean hasNullValue)
+ {
+ try {
+ if (type.equals(BOOLEAN)) {
+ boolean hasTrueValue = (boolean) minimum || (boolean) maximum;
+ boolean hasFalseValue = !(boolean) minimum || !(boolean) maximum;
+ if (hasTrueValue && hasFalseValue) {
+ return Domain.all(type);
+ }
+ if (hasTrueValue) {
+ return Domain.create(ValueSet.of(type, true), hasNullValue);
+ }
+ if (hasFalseValue) {
+ return Domain.create(ValueSet.of(type, false), hasNullValue);
+ }
+ // No other case, since all null case is handled earlier.
+ }
+
+ if ((type.equals(BIGINT) || type.equals(TINYINT) || type.equals(SMALLINT)
+ || type.equals(INTEGER) || type.equals(DATE))) {
+ long minValue = TupleDomainParquetPredicate.asLong(minimum);
+ long maxValue = TupleDomainParquetPredicate.asLong(maximum);
+ if (isStatisticsOverflow(type, minValue, maxValue)) {
+ return Domain.create(ValueSet.all(type), hasNullValue);
+ }
+ return ofMinMax(type, minValue, maxValue, hasNullValue);
+ }
+
+ if (type.equals(REAL)) {
+ Float minValue = (Float) minimum;
+ Float maxValue = (Float) maximum;
+ if (minValue.isNaN() || maxValue.isNaN()) {
+ return Domain.create(ValueSet.all(type), hasNullValue);
+ }
+ return ofMinMax(type, (long) floatToRawIntBits(minValue), (long) floatToRawIntBits(maxValue), hasNullValue);
+ }
+
+ if (type.equals(DOUBLE)) {
+ Double minValue = (Double) minimum;
+ Double maxValue = (Double) maximum;
+ if (minValue.isNaN() || maxValue.isNaN()) {
+ return Domain.create(ValueSet.all(type), hasNullValue);
+ }
+ return ofMinMax(type, minValue, maxValue, hasNullValue);
+ }
+
+ if (type.equals(VarcharType.VARCHAR)) {
+ Slice min = Slices.utf8Slice((String) minimum);
+ Slice max = Slices.utf8Slice((String) maximum);
+ return ofMinMax(type, min, max, hasNullValue);
+ }
+ return Domain.create(ValueSet.all(type), hasNullValue);
+ }
+ catch (Exception e) {
+ log.warn("failed to create Domain for column: %s which type is: %s", colName, type.toString());
+ return Domain.create(ValueSet.all(type), hasNullValue);
+ }
+ }
+
+ private static Domain ofMinMax(Type type, Object min, Object max, boolean hasNullValue)
+ {
+ Range range = Range.range(type, min, true, max, true);
+ ValueSet vs = ValueSet.ofRanges(ImmutableList.of(range));
+ return Domain.create(vs, hasNullValue);
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiDirectoryLister.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiDirectoryLister.java
new file mode 100644
index 0000000000000..cd961896af00d
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiDirectoryLister.java
@@ -0,0 +1,2 @@
+package com.facebook.presto.hudi.query.index;public class HudiDirectoryLister {
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiIndexSupport.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiIndexSupport.java
new file mode 100644
index 0000000000000..2fc1c9c6a4657
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiIndexSupport.java
@@ -0,0 +1,14 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.presto.common.predicate.TupleDomain;
+import org.apache.hudi.common.model.FileSlice;
+
+public interface HudiIndexSupport
+{
+ boolean canApply(TupleDomain tupleDomain);
+
+ default boolean shouldSkipFileSlice(FileSlice slice)
+ {
+ return false;
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiNoOpIndexSupport.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiNoOpIndexSupport.java
new file mode 100644
index 0000000000000..af059c45bb5b4
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiNoOpIndexSupport.java
@@ -0,0 +1,27 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.spi.SchemaTableName;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.util.Lazy;
+
+/**
+ * Noop index support to ensure that MDT enabled split generation is entered.
+ */
+public class HudiNoOpIndexSupport
+ extends HudiBaseIndexSupport
+{
+ private static final Logger log = Logger.get(HudiNoOpIndexSupport.class);
+
+ public HudiNoOpIndexSupport(SchemaTableName schemaTableName, Lazy lazyMetaClient)
+ {
+ super(log, schemaTableName, lazyMetaClient);
+ }
+
+ @Override
+ public boolean canApply(TupleDomain tupleDomain)
+ {
+ return true;
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiPartitionStatsIndexSupport.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiPartitionStatsIndexSupport.java
new file mode 100644
index 0000000000000..15e4dad0d92c0
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiPartitionStatsIndexSupport.java
@@ -0,0 +1,138 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.presto.common.predicate.Domain;
+import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.common.type.Type;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.SchemaTableName;
+import org.apache.hudi.avro.model.HoodieMetadataColumnStats;
+import org.apache.hudi.common.model.HoodieIndexDefinition;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.hash.ColumnIndexID;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.util.Lazy;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.facebook.presto.hudi.util.TupleDomainUtils.areSomeFieldsReferenced;
+import static com.facebook.presto.hudi.util.TupleDomainUtils.filter;
+import static com.facebook.presto.hudi.util.TupleDomainUtils.hasSimpleNullCheck;
+
+public class HudiPartitionStatsIndexSupport
+ extends HudiColumnStatsIndexSupport
+{
+ private static final Logger log = Logger.get(HudiColumnStatsIndexSupport.class);
+ private final Lazy lazyMetadataTable;
+
+ public HudiPartitionStatsIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy lazyMetaClient, Lazy lazyTableMetadata, TupleDomain regularColumnPredicates)
+ {
+ super(log, session, schemaTableName, lazyMetaClient, lazyTableMetadata, regularColumnPredicates);
+ this.lazyMetadataTable = lazyTableMetadata;
+ }
+
+ public Optional> prunePartitions(
+ Collection allPartitions)
+ {
+ HoodieTimer timer = HoodieTimer.start();
+
+ // Filter out predicates containing simple null checks (`IS NULL` or `IS NOT NULL`)
+ TupleDomain filteredRegularPredicates = filter(regularColumnPredicates, (ignored, domain) -> !hasSimpleNullCheck(domain));
+
+ // Sanity check, if no regular domains, return immediately
+ if (filteredRegularPredicates.getDomains().isEmpty()) {
+ timer.endTimer();
+ return Optional.empty();
+ }
+
+ List regularColumns = new ArrayList<>(filteredRegularPredicates.getDomains().get().keySet());
+
+ // Get columns to filter on
+ List encodedTargetColumnNames = regularColumns.stream()
+ .map(col -> new ColumnIndexID(col).asBase64EncodedString()).toList();
+
+ Map columnTypes = regularColumnPredicates.getDomains().get().entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getType()));
+
+ // Map of domains with partition stats keyed by partition name and column name
+ Map> domainsWithStats = lazyMetadataTable.get().getRecordsByKeyPrefixes(
+ encodedTargetColumnNames,
+ HoodieTableMetadataUtil.PARTITION_NAME_PARTITION_STATS, true)
+ .collectAsList()
+ .stream()
+ .filter(f -> f.getData().getColumnStatMetadata().isPresent())
+ .map(f -> f.getData().getColumnStatMetadata().get())
+ .collect(Collectors.groupingBy(
+ HoodieMetadataColumnStats::getFileName,
+ Collectors.toMap(
+ HoodieMetadataColumnStats::getColumnName,
+ // Pre-compute the Domain object for each HoodieMetadataColumnStats
+ stats -> getDomainFromColumnStats(stats.getColumnName(), columnTypes.get(stats.getColumnName()), stats))));
+
+ // For each partition, determine if it should be kept based on stats availability and predicate evaluation
+ Stream prunedPartitions = allPartitions.stream()
+ .filter(partition -> {
+ // Check if stats exist for this partition
+ Map partitionDomainsWithStats = domainsWithStats.get(partition);
+ if (partitionDomainsWithStats == null) {
+ // Partition has no stats in the index, keep it
+ return true;
+ }
+ else {
+ // Partition has stats, evaluate the predicate against them
+ // Keep the partition only if the predicate evaluates to true
+ // Important: If some columns in encodedTargetColumnNames is not available in partition stats, partition will not be pruned iff all available predicate
+ // evaluates to true. Since we cannot determine if the predicate will evaluate to true or not on the missing stat, adopt conservative measure to true,
+ // i.e. to not prune
+ return evaluateStatisticPredicate(filteredRegularPredicates, partitionDomainsWithStats, regularColumns);
+ }
+ });
+
+ log.info("Took %s ms to prune partitions using Partition Stats Index for table %s", timer.endTimer(), schemaTableName);
+ return Optional.of(prunedPartitions);
+ }
+
+ @Override
+ public boolean isIndexSupportAvailable()
+ {
+ return lazyMetaClient.get().getTableConfig().getMetadataPartitions()
+ .contains(HoodieTableMetadataUtil.PARTITION_NAME_PARTITION_STATS);
+ }
+
+ @Override
+ public boolean canApply(TupleDomain tupleDomain)
+ {
+ // Important: has the same implementation as col stats superclass, only difference is that log messages are different
+ if (!isIndexSupportAvailable()) {
+ log.debug("Partition Stats Index partition is not enabled in metadata table.");
+ return false;
+ }
+
+ Map indexDefinitions = getAllIndexDefinitions();
+ HoodieIndexDefinition partitionsStatsIndex = indexDefinitions.get(HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS);
+ if (partitionsStatsIndex == null || partitionsStatsIndex.getSourceFields() == null || partitionsStatsIndex.getSourceFields().isEmpty()) {
+ log.warn("Partition stats index definition is missing or has no source fields defined");
+ return false;
+ }
+
+ // Optimization applied: Only consider applicable if predicates reference indexed columns
+ List sourceFields = partitionsStatsIndex.getSourceFields();
+ boolean applicable = areSomeFieldsReferenced(tupleDomain, sourceFields);
+
+ if (applicable) {
+ log.debug("Partition Stats Index is available and applicable (predicates reference indexed columns).");
+ }
+ else {
+ log.debug("Partition Stats Index is available, but predicates do not reference any indexed columns.");
+ }
+ return applicable;
+ }
+}
\ No newline at end of file
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiRecordLevelIndexSupport.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiRecordLevelIndexSupport.java
new file mode 100644
index 0000000000000..0d852f8359c39
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiRecordLevelIndexSupport.java
@@ -0,0 +1,317 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.airlift.units.Duration;
+import com.facebook.presto.common.predicate.Domain;
+import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.SchemaTableName;
+import io.airlift.slice.Slice;
+import org.apache.hudi.common.data.HoodieListData;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.HoodieDataUtils;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.util.Lazy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+import static com.facebook.presto.hudi.HudiErrorCode.HUDI_BAD_DATA;
+import static com.facebook.presto.hudi.HudiSessionProperties.getRecordIndexWaitTimeout;
+import static com.facebook.presto.hudi.util.TupleDomainUtils.areAllFieldsReferenced;
+import static com.facebook.presto.hudi.util.TupleDomainUtils.areDomainsInOrEqualOnly;
+import static com.facebook.presto.hudi.util.TupleDomainUtils.getDiscreteSet;
+import static com.facebook.presto.hudi.util.TupleDomainUtils.isDiscreteSet;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+public class HudiRecordLevelIndexSupport
+ extends HudiBaseIndexSupport
+{
+ private static final Logger log = Logger.get(HudiRecordLevelIndexSupport.class);
+
+ public static final String DEFAULT_COLUMN_VALUE_SEPARATOR = ":";
+ public static final String DEFAULT_RECORD_KEY_PARTS_SEPARATOR = ",";
+ private final CompletableFuture>> relevantFileIdsFuture;
+ private final Duration recordIndexWaitTimeout;
+ private final long futureStartTimeMs;
+
+ public HudiRecordLevelIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy lazyMetaClient, Lazy lazyTableMetadata, TupleDomain regularColumnPredicates)
+ {
+ super(log, schemaTableName, lazyMetaClient);
+ this.recordIndexWaitTimeout = getRecordIndexWaitTimeout(session);
+ if (regularColumnPredicates.isAll()) {
+ log.debug("Predicates cover all data, skipping record level index lookup.");
+ this.relevantFileIdsFuture = CompletableFuture.completedFuture(Optional.empty());
+ }
+ else {
+ this.relevantFileIdsFuture = CompletableFuture.supplyAsync(() -> {
+ HoodieTimer timer = HoodieTimer.start();
+ Option recordKeyFieldsOpt = lazyMetaClient.get().getTableConfig().getRecordKeyFields();
+ if (recordKeyFieldsOpt.isEmpty() || recordKeyFieldsOpt.get().length == 0) {
+ // Should not happen since canApply checks for this, include for safety
+ throw new PrestoException(HUDI_BAD_DATA, "Record key fields must be defined to use Record Level Index.");
+ }
+ List recordKeyFields = Arrays.asList(recordKeyFieldsOpt.get());
+
+ // Only extract the predicates relevant to the record key fields
+ TupleDomain filteredDomains = extractPredicatesForColumns(regularColumnPredicates, recordKeyFields);
+
+ // Construct the actual record keys based on the filtered predicates using Hudi's encoding scheme
+ List recordKeys = constructRecordKeys(filteredDomains, recordKeyFields);
+
+ if (recordKeys.isEmpty()) {
+ // If key construction fails (e.g., incompatible predicates not caught by canApply, or placeholder issue)
+ log.warn("Took %s ms, but could not construct record keys from predicates. Skipping record index pruning for table %s",
+ timer.endTimer(), schemaTableName);
+ return Optional.empty();
+ }
+ log.debug(String.format("Constructed %d record keys for index lookup.", recordKeys.size()));
+
+ // Perform index lookup in metadataTable
+ // TODO: document here what this map is keyed by
+ Map recordIndex = HoodieDataUtils.dedupeAndCollectAsMap(lazyTableMetadata.get().readRecordIndexLocationsWithKeys(HoodieListData.lazy(recordKeys)));
+ if (recordIndex.isEmpty()) {
+ log.debug("Record level index lookup took %s ms but returned no locations for the given keys %s for table %s",
+ timer.endTimer(), recordKeys, schemaTableName);
+ // Return all original fileSlices
+ return Optional.empty();
+ }
+
+ // Collect fileIds for pruning
+ Set relevantFiles = recordIndex.values().stream()
+ .map(HoodieRecordGlobalLocation::getFileId)
+ .collect(Collectors.toSet());
+ log.debug("Record level index lookup took %s ms and identified %d relevant file IDs.", timer.endTimer(), relevantFiles.size());
+
+ return Optional.of(relevantFiles);
+ });
+ }
+ this.futureStartTimeMs = System.currentTimeMillis();
+ }
+
+ @Override
+ public boolean shouldSkipFileSlice(FileSlice slice)
+ {
+ try {
+ if (relevantFileIdsFuture.isDone()) {
+ Optional> relevantFileIds = relevantFileIdsFuture.get();
+ return relevantFileIds.map(fileIds -> !fileIds.contains(slice.getFileId())).orElse(false);
+ }
+
+ long elapsedMs = System.currentTimeMillis() - futureStartTimeMs;
+ if (elapsedMs > recordIndexWaitTimeout.toMillis()) {
+ // Took too long; skip decision
+ return false;
+ }
+
+ long remainingMs = Math.max(0, recordIndexWaitTimeout.toMillis() - elapsedMs);
+ Optional> relevantFileIds = relevantFileIdsFuture.get(remainingMs, MILLISECONDS);
+ return relevantFileIds.map(fileIds -> !fileIds.contains(slice.getFileId())).orElse(false);
+ }
+ catch (TimeoutException | InterruptedException | ExecutionException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Checks if the Record Level Index is available and the query predicates
+ * reference all record key fields with compatible (IN/EQUAL) constraints.
+ */
+ @Override
+ public boolean canApply(TupleDomain tupleDomain)
+ {
+ if (!isIndexSupportAvailable()) {
+ log.debug("Record Level Index partition is not enabled in metadata.");
+ return false;
+ }
+
+ Option recordKeyFieldsOpt = lazyMetaClient.get().getTableConfig().getRecordKeyFields();
+ if (recordKeyFieldsOpt.isEmpty() || recordKeyFieldsOpt.get().length == 0) {
+ log.debug("Record key fields are not defined in table config.");
+ return false;
+ }
+ List recordKeyFields = Arrays.asList(recordKeyFieldsOpt.get());
+
+ // Ensure that predicates reference all record key fields and use IN/EQUAL
+ boolean applicable = areAllFieldsReferenced(tupleDomain, recordKeyFields)
+ && areDomainsInOrEqualOnly(tupleDomain, recordKeyFields);
+
+ if (!applicable) {
+ log.debug("Predicates do not reference all record key fields or use non-compatible (non IN/EQUAL) constraints.");
+ }
+ else {
+ log.debug("Record Level Index is available and applicable based on predicates.");
+ }
+ return applicable;
+ }
+
+ private boolean isIndexSupportAvailable()
+ {
+ return lazyMetaClient.get().getTableConfig().getMetadataPartitions()
+ .contains(HoodieTableMetadataUtil.PARTITION_NAME_RECORD_INDEX);
+ }
+
+ /**
+ * Extracts predicates from a TupleDomain that match a given set of columns.
+ * Preserves all complex predicate properties including multi-value domains,
+ * range-based predicates, and nullability.
+ *
+ * @param tupleDomain The source TupleDomain containing all predicates
+ * @param columnFields The set of columns for which to extract predicates
+ * @return A new TupleDomain containing only the predicates for the specified columns
+ */
+ public static TupleDomain extractPredicatesForColumns(TupleDomain tupleDomain, List columnFields)
+ {
+ if (tupleDomain.isNone()) {
+ return TupleDomain.none();
+ }
+
+ if (tupleDomain.isAll()) {
+ return TupleDomain.all();
+ }
+
+ // Extract the domains matching the specified columns
+ Map allDomains = tupleDomain.getDomains().get();
+ Map filteredDomains = allDomains.entrySet().stream().filter(entry -> columnFields.contains(entry.getKey())) // Ensure key is in the column set
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ // If no domains matched, but we had some columns to extract, return ALL
+ if (filteredDomains.isEmpty() && !columnFields.isEmpty() && !allDomains.isEmpty()) {
+ return TupleDomain.all();
+ }
+
+ return TupleDomain.withColumnDomains(filteredDomains);
+ }
+
+ /**
+ * Constructs a record key from TupleDomain based on whether it's a complex key or not.
+ *
+ * Construction of record keys will only be handled for domains generated from EQUALITY or IN predicates.
+ *
+ * An empty list of record keys will be generated if the following conditions are not met:
+ *
+ * - recordKeysFields is empty
+ * - recordKeyDomains isAll
+ * - For the case of complex key, domains are not applied to all recordKeysFields
+ * - For the case of complex key, domains are applied to all recordKeyFields, but one of the domain is NOT
+ * generated from an equality or IN predicate
+ *
+ *
+ * Note: This function is O(m^n) where m is the average size of value literals and n is the number of record keys.
+ *
+ * Optimization 1: If MDT enabled functions allows for streams to be passed in, we can implement an iterator to be more memory efficient.
+ *
+ * Optimization 2: We should also consider limiting the number of recordKeys generated, if it is estimated to be more than a limit, RLI should just be skipped
+ * as it may just be faster to read out all data and filer accordingly.
+ *
+ * @param recordKeyDomains The filtered TupleDomain containing column handles and values
+ * @param recordKeyFields List of column names that represent the record keys
+ * @return List of string values representing the record key(s)
+ */
+ public static List constructRecordKeys(TupleDomain recordKeyDomains, List recordKeyFields)
+ {
+ // TODO: Move this to TupleDomainUtils
+ // If no recordKeys or no recordKeyDomains, return empty list
+ if (recordKeyFields == null || recordKeyFields.isEmpty() || recordKeyDomains.isAll()) {
+ return Collections.emptyList();
+ }
+
+ // All recordKeys must have a domain else, return empty list (applicable to complexKeys)
+ // If a one of the recordKey in the set of complexKeys does not have a domain, we are unable to construct
+ // a complete complexKey
+ if (!recordKeyDomains.getDomains().get().keySet().containsAll(recordKeyFields)) {
+ return Collections.emptyList();
+ }
+
+ // Extract the domain mappings from the tuple domain
+ Map domains = recordKeyDomains.getDomains().get();
+
+ // Case 1: Not a complex key (single record key)
+ if (recordKeyFields.size() == 1) {
+ String recordKey = recordKeyFields.get(0);
+
+ // Extract value for this key
+ Domain domain = domains.get(recordKey);
+ return extractStringValues(domain);
+ }
+ // Case 2: Complex/Composite key (multiple record keys)
+ else {
+ // Create a queue to manage the Cartesian product generation
+ Queue results = new LinkedList<>();
+
+ // For each key in the complex key
+ for (String recordKeyField : recordKeyFields) {
+ // Extract value for this key
+ Domain domain = domains.get(recordKeyField);
+ List values = extractStringValues(domain);
+ // First iteration: initialize the queue
+ if (results.isEmpty()) {
+ values.forEach(v -> results.offer(recordKeyField + DEFAULT_COLUMN_VALUE_SEPARATOR + v));
+ }
+ else {
+ int size = results.size();
+ for (int j = 0; j < size; j++) {
+ String currentEntry = results.poll();
+
+ // Generate new combinations by appending keyParts to existing keyParts
+ for (String v : values) {
+ String newKeyPart = recordKeyField + DEFAULT_COLUMN_VALUE_SEPARATOR + v;
+ String newEntry = currentEntry + DEFAULT_RECORD_KEY_PARTS_SEPARATOR + newKeyPart;
+ results.offer(newEntry);
+ }
+ }
+ }
+ }
+ return results.stream().toList();
+ }
+ }
+
+ /**
+ * Extract string values from a domain, handle EQUAL and IN domains only.
+ * Note: Actual implementation depends on your Domain class structure.
+ */
+ private static List extractStringValues(Domain domain)
+ {
+ List values = new ArrayList<>();
+
+ if (domain.isSingleValue()) {
+ // Handle EQUAL condition (single value domain)
+ Object value = domain.getSingleValue();
+ values.add(convertToString(value));
+ }
+ else if (isDiscreteSet(domain.getValues())) {
+ // Handle IN condition (set of discrete values)
+ for (Object value : getDiscreteSet(domain.getValues())) {
+ values.add(convertToString(value));
+ }
+ }
+ return values;
+ }
+
+ private static String convertToString(Object value)
+ {
+ if (value instanceof Slice) {
+ return ((Slice) value).toStringUtf8();
+ }
+ else {
+ return value.toString();
+ }
+ }
+}
\ No newline at end of file
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiSecondaryIndexSupport.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiSecondaryIndexSupport.java
new file mode 100644
index 0000000000000..0b106d316ad5c
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/HudiSecondaryIndexSupport.java
@@ -0,0 +1,221 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.airlift.units.Duration;
+import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.hudi.util.TupleDomainUtils;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.SchemaTableName;
+import org.apache.hudi.common.data.HoodieListData;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieIndexDefinition;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.HoodieDataUtils;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.util.Lazy;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+import static com.facebook.presto.hudi.HudiSessionProperties.getSecondaryIndexWaitTimeout;
+import static com.facebook.presto.hudi.query.index.HudiRecordLevelIndexSupport.constructRecordKeys;
+import static com.facebook.presto.hudi.query.index.HudiRecordLevelIndexSupport.extractPredicatesForColumns;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+public class HudiSecondaryIndexSupport
+ extends HudiBaseIndexSupport
+{
+ private static final Logger log = Logger.get(HudiSecondaryIndexSupport.class);
+ private final CompletableFuture>> relevantFileIdsFuture;
+ private final Duration secondaryIndexWaitTimeout;
+ private final long futureStartTimeMs;
+
+ public HudiSecondaryIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy lazyMetaClient, Lazy lazyTableMetadata, TupleDomain regularColumnPredicates)
+ {
+ super(log, schemaTableName, lazyMetaClient);
+ this.secondaryIndexWaitTimeout = getSecondaryIndexWaitTimeout(session);
+ this.relevantFileIdsFuture = CompletableFuture.supplyAsync(() -> {
+ HoodieTimer timer = HoodieTimer.start();
+ if (regularColumnPredicates.isAll() || lazyMetaClient.get().getIndexMetadata().isEmpty()) {
+ log.debug("Predicates cover all data, skipping secondary index lookup.");
+ return Optional.empty();
+ }
+
+ Optional> firstApplicableIndex = findFirstApplicableSecondaryIndex(regularColumnPredicates);
+ if (firstApplicableIndex.isEmpty()) {
+ log.debug("Took %s ms but no secondary index definition found matching the query's referenced columns for table %s",
+ timer.endTimer(), schemaTableName);
+ return Optional.empty();
+ }
+
+ Map.Entry applicableIndexEntry = firstApplicableIndex.get();
+ String indexName = applicableIndexEntry.getKey();
+ // `indexedColumns` should only contain one element as secondary indices only support one column
+ List indexedColumns = applicableIndexEntry.getValue().getSourceFields();
+ log.debug(String.format("Using secondary index '%s' on columns %s for pruning.", indexName, indexedColumns));
+ TupleDomain indexPredicates = extractPredicatesForColumns(regularColumnPredicates, indexedColumns);
+
+ List secondaryKeys = constructRecordKeys(indexPredicates, indexedColumns);
+ if (secondaryKeys.isEmpty()) {
+ log.warn("Took %s ms, but could not construct secondary keys for index '%s' from predicates. Skipping pruning for table %s",
+ timer.endTimer(), indexName, schemaTableName);
+ return Optional.empty();
+ }
+ log.debug(String.format("Constructed %d secondary keys for index lookup.", secondaryKeys.size()));
+
+ // Perform index lookup in metadataTable
+ // TODO: document here what this map is keyed by
+ Map recordKeyLocationsMap =
+ HoodieDataUtils.dedupeAndCollectAsMap(lazyTableMetadata.get().readSecondaryIndexLocationsWithKeys(
+ HoodieListData.lazy(secondaryKeys), indexName));
+ if (recordKeyLocationsMap.isEmpty()) {
+ log.debug("Took %s ms, but secondary index lookup returned no locations for the given keys for table %s", timer.endTimer(), schemaTableName);
+ // Return all original fileSlices
+ return Optional.empty();
+ }
+
+ // Collect fileIds for pruning
+ Set relevantFileIds = recordKeyLocationsMap.values().stream()
+ .map(HoodieRecordGlobalLocation::getFileId)
+ .collect(Collectors.toSet());
+ log.debug(String.format("Secondary index lookup identified %d relevant file IDs.", relevantFileIds.size()));
+
+ return Optional.of(relevantFileIds);
+ });
+
+ this.futureStartTimeMs = System.currentTimeMillis();
+ }
+
+ @Override
+ public boolean shouldSkipFileSlice(FileSlice slice)
+ {
+ try {
+ if (relevantFileIdsFuture.isDone()) {
+ Optional> relevantFileIds = relevantFileIdsFuture.get();
+ return relevantFileIds.map(fileIds -> !fileIds.contains(slice.getFileId())).orElse(false);
+ }
+
+ long elapsedMs = System.currentTimeMillis() - futureStartTimeMs;
+ if (elapsedMs > secondaryIndexWaitTimeout.toMillis()) {
+ // Took too long; skip decision
+ return false;
+ }
+
+ long remainingMs = Math.max(0, secondaryIndexWaitTimeout.toMillis() - elapsedMs);
+ Optional> relevantFileIds = relevantFileIdsFuture.get(remainingMs, MILLISECONDS);
+ return relevantFileIds.map(fileIds -> !fileIds.contains(slice.getFileId())).orElse(false);
+ }
+ catch (TimeoutException | InterruptedException | ExecutionException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Determines whether secondary index (SI) should be used based on the given tuple domain and index definitions.
+ *
+ * This method first filters out the secondary index definitions from the provided map of index definitions.
+ * It then checks if there are any secondary indices defined. If no secondary indices are found, it returns {@code false}.
+ *
+ * For each secondary index definition, the method verifies two conditions:
+ *
+ * - All fields referenced in the tuple domain must be part of the source fields of the secondary index.
+ * - The predicates on these fields must be either of type IN or EQUAL.
+ *
+ *
+ * If at least one secondary index definition meets these conditions, the method returns {@code true}; otherwise,
+ * it returns {@code false}.
+ *
+ * @param tupleDomain the domain representing the constraints on the columns
+ * HoodieIndexDefinition object
+ * @return {@code true} if at least one secondary index can be used based on the given tuple domain; otherwise,
+ * {@code false}
+ */
+ @Override
+ public boolean canApply(TupleDomain tupleDomain)
+ {
+ if (!isIndexSupportAvailable()) {
+ log.debug("Secondary Index partition is not enabled in metadata.");
+ return false;
+ }
+
+ Map secondaryIndexDefinitions = getApplicableIndexDefinitions(tupleDomain, true);
+ if (secondaryIndexDefinitions.isEmpty()) {
+ log.debug("No applicable secondary index definitions found.");
+ return false;
+ }
+
+ boolean atLeastOneIndexUsable = secondaryIndexDefinitions.values().stream()
+ .anyMatch(indexDef -> {
+ List sourceFields = indexDef.getSourceFields();
+ // Predicates referencing columns with secondary index needs to be IN or EQUAL only
+ boolean usable = TupleDomainUtils.areAllFieldsReferenced(tupleDomain, sourceFields)
+ && TupleDomainUtils.areDomainsInOrEqualOnly(tupleDomain, sourceFields);
+ if (log.isDebugEnabled() && usable) {
+ log.debug(String.format("Secondary index '%s' on fields '%s' is usable for the query.", indexDef.getIndexName(), sourceFields));
+ }
+ return usable;
+ });
+ if (!atLeastOneIndexUsable) {
+ log.debug("Although secondary indexes exist, none match the required fields and predicate types (IN/EQUAL) for the query.");
+ }
+ return atLeastOneIndexUsable;
+ }
+
+ private boolean isIndexSupportAvailable()
+ {
+ // Filter out definitions that are secondary indices
+ Map secondaryIndexDefinitions = getAllIndexDefinitions()
+ .entrySet().stream()
+ .filter(e -> e.getKey().contains(HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX))
+ .collect(Collectors.toMap(e -> e.getValue().getIndexName(),
+ Map.Entry::getValue));
+ return !secondaryIndexDefinitions.isEmpty();
+ }
+
+ private Map getApplicableIndexDefinitions(TupleDomain tupleDomain, boolean checkPredicateCompatibility)
+ {
+ Map allDefinitions = getAllIndexDefinitions();
+ if (allDefinitions.isEmpty()) {
+ return Map.of();
+ }
+ // Filter out definitions that are secondary indices
+ return allDefinitions.entrySet().stream()
+ .filter(entry -> entry.getKey().contains(HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX))
+ .filter(entry -> {
+ if (!checkPredicateCompatibility) {
+ return true;
+ }
+ // Perform additional compatibility checks
+ List sourceFields = entry.getValue().getSourceFields();
+ return TupleDomainUtils.areAllFieldsReferenced(tupleDomain, sourceFields)
+ && TupleDomainUtils.areDomainsInOrEqualOnly(tupleDomain, sourceFields);
+ })
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ }
+
+ private Optional> findFirstApplicableSecondaryIndex(TupleDomain queryPredicates)
+ {
+ // Predicate checks would have already been done, skip predicate checks here
+ Map secondaryIndexDefinitions = getApplicableIndexDefinitions(queryPredicates, false);
+ if (queryPredicates.getDomains().isEmpty()) {
+ return Optional.empty();
+ }
+ List queryReferencedColumns = List.copyOf(queryPredicates.getDomains().get().keySet());
+ return secondaryIndexDefinitions.entrySet().stream()
+ .filter(entry -> {
+ List sourceFields = entry.getValue().getSourceFields();
+ // Only filter for sourceFields that match the predicates
+ return !sourceFields.isEmpty() && queryReferencedColumns.contains(sourceFields.get(0));
+ })
+ .findFirst();
+ }
+}
\ No newline at end of file
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/IndexSupportFactory.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/IndexSupportFactory.java
new file mode 100644
index 0000000000000..7cb33864f4932
--- /dev/null
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/query/index/IndexSupportFactory.java
@@ -0,0 +1,152 @@
+package com.facebook.presto.hudi.query.index;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.hudi.HudiColumnHandle;
+import com.facebook.presto.hudi.HudiTableHandle;
+import com.facebook.presto.hudi.HudiTableLayoutHandle;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.SchemaTableName;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.util.Lazy;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import static com.facebook.presto.hudi.HudiSessionProperties.isColumnStatsIndexEnabled;
+import static com.facebook.presto.hudi.HudiSessionProperties.isNoOpIndexEnabled;
+import static com.facebook.presto.hudi.HudiSessionProperties.isPartitionStatsIndexEnabled;
+import static com.facebook.presto.hudi.HudiSessionProperties.isRecordLevelIndexEnabled;
+import static com.facebook.presto.hudi.HudiSessionProperties.isResolveColumnNameCasingEnabled;
+import static com.facebook.presto.hudi.HudiSessionProperties.isSecondaryIndexEnabled;
+import static com.facebook.presto.hudi.util.HudiUtil.getFieldFromSchema;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Factory to create the appropriate HudiIndexSupport strategy based on:
+ * 1. Query predicates
+ * 2. Available table indexes
+ * 3. Configuration flags
+ */
+public class IndexSupportFactory
+{
+ private static final Logger log = Logger.get(IndexSupportFactory.class);
+
+ private IndexSupportFactory() {}
+
+ /**
+ * Creates the most suitable HudiIndexSupport strategy, considering configuration.
+ * Uses Supplier-based lazy instantiation combined with config checks.
+ *
+ * @param layoutHandle The hudi table layout handle
+ * @param lazyMetaClient The Hudi table metadata client that is lazily instantiated.
+ * @param tupleDomain The query predicates.
+ * @param session Session containing session properties, which is required to control index behaviours for testing/debugging
+ * @return An Optional containing the chosen HudiIndexSupport strategy, or empty if none are applicable or enabled.
+ */
+ public static Optional createIndexSupport(
+ HudiTableLayoutHandle layoutHandle,
+ Lazy lazyMetaClient,
+ Lazy lazyTableMetadata,
+ TupleDomain tupleDomain,
+ ConnectorSession session)
+ {
+ TupleDomain transformedTupleDomain = transformTupleDomain(session, layoutHandle, tupleDomain);
+ HudiTableHandle hudiTableHandle = layoutHandle.getTableHandle();
+ SchemaTableName schemaTableName = hudiTableHandle.getSchemaTableName();
+ // Define strategies as Suppliers paired with their config (isEnabled) flag
+ // IMPORTANT: Order of strategy here determines which index implementation is preferred first
+ List strategyProviders = List.of(
+ new StrategyProvider(
+ () -> isRecordLevelIndexEnabled(session),
+ () -> new HudiRecordLevelIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, transformedTupleDomain)),
+ new StrategyProvider(
+ () -> isSecondaryIndexEnabled(session),
+ () -> new HudiSecondaryIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, transformedTupleDomain)),
+ new StrategyProvider(
+ () -> isColumnStatsIndexEnabled(session),
+ () -> new HudiColumnStatsIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, transformedTupleDomain)),
+ new StrategyProvider(
+ () -> isNoOpIndexEnabled(session),
+ () -> new HudiNoOpIndexSupport(schemaTableName, lazyMetaClient)));
+
+ for (StrategyProvider provider : strategyProviders) {
+ // Check if the strategy is enabled via config before instantiating
+ if (provider.isEnabled()) {
+ HudiIndexSupport strategy = provider.getStrategy();
+ String strategyName = strategy.getClass().getSimpleName(); // Get name for logging
+
+ // Check if the instantiated strategy is applicable
+ if (strategy.canApply(transformedTupleDomain)) {
+ log.debug(String.format("Selected %s strategy (Enabled & Applicable).", strategyName));
+ return Optional.of(strategy);
+ }
+ else {
+ log.debug(String.format("%s is enabled but not applicable for this query.", strategyName));
+ // Strategy object becomes eligible for GC here, acceptable penalty as the object is lightweight
+ }
+ }
+ else {
+ log.debug(String.format("Strategy associated with supplier %s is disabled by configuration.", provider.supplier.get().getClass().getSimpleName()));
+ }
+ }
+
+ log.debug("No suitable and enabled index support strategy found to be applicable.");
+ return Optional.empty();
+ }
+
+ public static Optional createPartitionStatsIndexSupport(
+ HudiTableHandle hudiTableHandle,
+ Lazy lazyMetaClient,
+ Lazy lazyTableMetadata,
+ TupleDomain tupleDomain,
+ ConnectorSession session)
+ {
+ TupleDomain transformedTupleDomain = tupleDomain.transform(HudiColumnHandle::getName);;;
+
+ StrategyProvider partitionStatsStrategy = new StrategyProvider(
+ () -> isPartitionStatsIndexEnabled(session), () -> new HudiPartitionStatsIndexSupport(session, hudiTableHandle.getSchemaTableName(), lazyMetaClient, lazyTableMetadata, transformedTupleDomain));
+
+ if (partitionStatsStrategy.isEnabled() && partitionStatsStrategy.getStrategy().canApply(transformedTupleDomain)) {
+ return Optional.of((HudiPartitionStatsIndexSupport) partitionStatsStrategy.getStrategy());
+ }
+ return Optional.empty();
+ }
+
+ private static TupleDomain transformTupleDomain(ConnectorSession session, HudiTableLayoutHandle layoutHandle, TupleDomain tupleDomain)
+ {
+ if (isResolveColumnNameCasingEnabled(session)) {
+ // if column case reconciliation is enabled, transform the tuple domain keys to match the column names from the Hudi table.
+ return tupleDomain.transform(hiveColumnHandle ->
+ getFieldFromSchema(hiveColumnHandle.getName(), layoutHandle.getTableSchema()).name());
+ }
+ return tupleDomain.transform(HudiColumnHandle::getName);
+ }
+
+ /**
+ * Helper class to pair the configuration check with the strategy supplier to allow for lazy initialization.
+ */
+ private static class StrategyProvider
+ {
+ private final Supplier isEnabled;
+ private final Supplier supplier;
+
+ StrategyProvider(Supplier isEnabled, Supplier supplier)
+ {
+ this.isEnabled = requireNonNull(isEnabled);
+ this.supplier = requireNonNull(supplier);
+ }
+
+ boolean isEnabled()
+ {
+ return isEnabled.get();
+ }
+
+ HudiIndexSupport getStrategy()
+ {
+ return supplier.get();
+ }
+ }
+}
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/split/HudiBackgroundSplitLoader.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/split/HudiBackgroundSplitLoader.java
index 14894adf26318..987151c8a8615 100644
--- a/presto-hudi/src/main/java/com/facebook/presto/hudi/split/HudiBackgroundSplitLoader.java
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/split/HudiBackgroundSplitLoader.java
@@ -14,25 +14,69 @@
package com.facebook.presto.hudi.split;
+import com.facebook.airlift.concurrent.BoundedExecutor;
import com.facebook.airlift.log.Logger;
-import com.facebook.presto.hive.metastore.ExtendedHiveMetastore;
+import com.facebook.presto.hive.metastore.Column;
+import com.facebook.presto.hive.metastore.Partition;
+import com.facebook.presto.hive.metastore.StorageFormat;
+import com.facebook.presto.hive.metastore.Table;
import com.facebook.presto.hive.util.AsyncQueue;
+import com.facebook.presto.hudi.HudiColumnHandle;
+import com.facebook.presto.hudi.HudiPartition;
+import com.facebook.presto.hudi.HudiTableHandle;
import com.facebook.presto.hudi.HudiTableLayoutHandle;
+import com.facebook.presto.hudi.partition.HiveStylePartitionValueExtractor;
+import com.facebook.presto.hudi.partition.MultiPartKeysValueExtractor;
+import com.facebook.presto.hudi.partition.NonPartitionedExtractor;
+import com.facebook.presto.hudi.partition.PartitionValueExtractor;
+import com.facebook.presto.hudi.partition.SinglePartPartitionValueExtractor;
+import com.facebook.presto.hudi.query.HudiDirectoryLister;
+import com.facebook.presto.hudi.query.index.HudiPartitionStatsIndexSupport;
+import com.facebook.presto.hudi.query.index.IndexSupportFactory;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.ConnectorSplit;
import com.facebook.presto.spi.PrestoException;
-import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
-import org.apache.hudi.common.util.HoodieTimer;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Streams;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.util.Lazy;
import java.util.ArrayList;
+import java.util.Deque;
+import java.util.HexFormat;
import java.util.List;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
-import static com.facebook.presto.hudi.HudiErrorCode.HUDI_CANNOT_GENERATE_SPLIT;
+import static com.facebook.airlift.concurrent.MoreFutures.addExceptionCallback;
+import static com.facebook.presto.hive.metastore.MetastoreUtil.extractPartitionValues;
+import static com.facebook.presto.hudi.HudiErrorCode.HUDI_CANNOT_OPEN_SPLIT;
+import static com.facebook.presto.hudi.HudiErrorCode.HUDI_INVALID_METADATA;
+import static com.facebook.presto.hudi.HudiMetadata.fromDataColumns;
import static com.facebook.presto.hudi.HudiSessionProperties.getSplitGeneratorParallelism;
+import static com.facebook.presto.hudi.HudiSessionProperties.isMetadataPartitionListingEnabled;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
/**
@@ -42,65 +86,294 @@ public class HudiBackgroundSplitLoader
implements Runnable
{
private static final Logger log = Logger.get(HudiBackgroundSplitLoader.class);
+ public static final String DELIMITER_STR = "/";
+ public static final String EQUALS_STR = "=";
private final ConnectorSession session;
- private final ExtendedHiveMetastore metastore;
private final HudiTableLayoutHandle layout;
- private final HoodieTableFileSystemView fsView;
+ private final HudiDirectoryLister hudiDirectoryLister;
private final AsyncQueue asyncQueue;
- private final List partitions;
- private final String latestInstant;
+ private final Lazy> lazyPartitionMap;
private final int splitGeneratorNumThreads;
private final ExecutorService splitGeneratorExecutorService;
+ private final boolean enableMetadataTable;
+ private final Lazy lazyTableMetadata;
+ private final Consumer errorListener;
+ private final Optional partitionIndexSupportOpt;
+ private final boolean isMetadataPartitionListingEnabled;
public HudiBackgroundSplitLoader(
ConnectorSession session,
- ExtendedHiveMetastore metastore,
ExecutorService splitGeneratorExecutorService,
HudiTableLayoutHandle layout,
- HoodieTableFileSystemView fsView,
+ HudiDirectoryLister hudiDirectoryLister,
AsyncQueue asyncQueue,
- List partitions,
- String latestInstant)
+ Lazy> lazyPartitionMap,
+ boolean enableMetadataTable,
+ Lazy lazyTableMetadata,
+ Consumer