This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 6b3f817c9ec [HUDI-8232] Adding Index defn for Col stats to track the
list of columns indexed (#12529)
6b3f817c9ec is described below
commit 6b3f817c9ecb5a1dc6922a61b4a29cae63f85333
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Fri Jan 3 21:31:43 2025 -0800
[HUDI-8232] Adding Index defn for Col stats to track the list of columns
indexed (#12529)
Adding Index definition for column stats to track the list of columns
indexed.
This will be later leveraged during pruning to lookup in
column_stats partition for indexed columns only.
---
.../org/apache/hudi/client/BaseHoodieClient.java | 8 ++
.../hudi/client/BaseHoodieTableServiceClient.java | 4 +
.../apache/hudi/client/BaseHoodieWriteClient.java | 7 +
.../apache/hudi/client/CompactionAdminClient.java | 5 +
.../hudi/client/HoodieColumnStatsIndexUtils.java | 87 ++++++++++++
.../metadata/HoodieBackedTableMetadataWriter.java | 37 ++++-
.../java/org/apache/hudi/table/HoodieTable.java | 5 +
.../action/commit/BaseCommitActionExecutor.java | 16 +++
.../index/functional/BaseHoodieIndexClient.java | 23 +++-
.../apache/hudi/client/HoodieFlinkWriteClient.java | 5 +
.../FlinkHoodieBackedTableMetadataWriter.java | 5 +
.../commit/BaseFlinkCommitActionExecutor.java | 7 +
.../hudi/client/HoodieJavaTableServiceClient.java | 6 +
.../apache/hudi/client/HoodieJavaWriteClient.java | 5 +
.../JavaHoodieBackedTableMetadataWriter.java | 5 +
.../commit/BaseJavaCommitActionExecutor.java | 5 +
.../hudi/testutils/TestHoodieMetadataBase.java | 1 +
.../org/apache/hudi/HoodieSparkIndexClient.java | 105 +++++++++++---
.../apache/hudi/client/SparkRDDWriteClient.java | 6 +
.../SparkHoodieBackedTableMetadataWriter.java | 6 +
.../SparkBootstrapCommitActionExecutor.java | 6 +
.../commit/BaseSparkCommitActionExecutor.java | 7 +
.../org/apache/spark/sql/HoodieUnsafeUtils.scala | 3 +-
.../apache/hudi/client/TestHoodieSparkIndex.java | 153 +++++++++++++++++++++
.../hudi/common/model/HoodieIndexDefinition.java | 20 +++
.../hudi/common/table/HoodieTableMetaClient.java | 45 ++++--
.../hudi/metadata/HoodieTableMetadataUtil.java | 70 +++++-----
.../org/apache/hudi/avro/TestHoodieAvroUtils.java | 2 +-
.../hudi/metadata/TestHoodieTableMetadataUtil.java | 16 ++-
.../cow-updated3-column-stats-index-table.json | 14 ++
.../cow-updated4-column-stats-index-table.json | 15 ++
.../mor-updated3-column-stats-index-table.json | 14 ++
.../mor-updated4-column-stats-index-table.json | 15 ++
...-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json | 1 +
...-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json | 1 +
.../hudi/functional/ColumnStatIndexTestBase.scala | 37 ++---
.../hudi/functional/TestColumnStatsIndex.scala | 82 ++++++++++-
.../functional/TestColumnStatsIndexWithSQL.scala | 17 ++-
38 files changed, 764 insertions(+), 102 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
index 26a5b191377..b744e4fd87d 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
@@ -284,4 +284,12 @@ public abstract class BaseHoodieClient implements
Serializable, AutoCloseable {
}
}
}
+
+ /**
+ * Updates the cols being indexed with column stats. This is for tracking
purpose so that queries can leverage col stats
+ * from MDT only for indexed columns.
+ * @param metaClient instance of {@link HoodieTableMetaClient} of interest.
+ * @param columnsToIndex list of columns to index.
+ */
+ protected abstract void
updateColumnsToIndexWithColStats(HoodieTableMetaClient metaClient, List<String>
columnsToIndex);
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
index 81a13e308e3..a5beda7c913 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
@@ -1197,6 +1197,10 @@ public abstract class BaseHoodieTableServiceClient<I, T,
O> extends BaseHoodieCl
}
}
+ protected void updateColumnsToIndexWithColStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ // no op
+ }
+
/**
* Called after each commit of a compaction or clustering table service,
* to release any resources used.
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index c58181eab45..bce127cc7b0 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -56,6 +56,7 @@ import org.apache.hudi.common.table.timeline.TimelineUtils;
import org.apache.hudi.common.util.CleanerUtils;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CommitUtils;
+import org.apache.hudi.common.util.Functions;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
@@ -286,6 +287,12 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
writeTableMetadata(table, instantTime, metadata);
activeTimeline.saveAsComplete(false,
table.getMetaClient().createNewInstant(HoodieInstant.State.INFLIGHT,
commitActionType, instantTime),
serializeCommitMetadata(table.getMetaClient().getCommitMetadataSerDe(),
metadata));
+ // update cols to Index as applicable
+ HoodieColumnStatsIndexUtils.updateColsToIndex(table, config, metadata,
+ (Functions.Function2<HoodieTableMetaClient, List<String>, Void>)
(metaClient, columnsToIndex) -> {
+ updateColumnsToIndexWithColStats(metaClient, columnsToIndex);
+ return null;
+ });
}
// Save internal schema
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
index 76bb49734ae..b7736c6b2f2 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
@@ -324,6 +324,11 @@ public class CompactionAdminClient extends
BaseHoodieClient {
}
}
+ @Override
+ protected void updateColumnsToIndexWithColStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ // no op
+ }
+
/**
* Holds Operation result for Renaming.
*/
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieColumnStatsIndexUtils.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieColumnStatsIndexUtils.java
new file mode 100644
index 00000000000..6b2feff15ec
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieColumnStatsIndexUtils.java
@@ -0,0 +1,87 @@
+/*
+ * 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.hudi.client;
+
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.Functions;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.table.HoodieTable;
+
+import java.util.List;
+
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
+
+/**
+ * Utils to assist with updating columns to index with col stats.
+ */
+public class HoodieColumnStatsIndexUtils {
+
+ /**
+ * Updates the list of columns to index with col stats partition in MDT.
+ * @param dataTable {@link HoodieTable} of interest.
+ * @param config {@link HoodieWriteConfig} of interest.
+ * @param commitMetadata commit metadata of interest.
+ * @param updateColSatsFunc function to assist with updating columns to
index.
+ */
+ @VisibleForTesting
+ public static void updateColsToIndex(HoodieTable dataTable,
HoodieWriteConfig config, HoodieCommitMetadata commitMetadata,
+ Functions.Function2<HoodieTableMetaClient,
List<String>, Void> updateColSatsFunc) {
+ if (config.getMetadataConfig().isColumnStatsIndexEnabled()) {
+ dataTable.getMetaClient().reloadTableConfig();
+ if
(dataTable.getMetaClient().getTableConfig().getMetadataPartitions().contains(MetadataPartitionType.COLUMN_STATS.getPartitionPath()))
{
+ try {
+ HoodieTableMetaClient mdtMetaClient = HoodieTableMetaClient.builder()
+ .setStorage(dataTable.getStorage())
+
.setBasePath(HoodieTableMetadata.getMetadataTableBasePath(dataTable.getMetaClient().getBasePath()))
+ .build();
+ HoodieInstant latestInstant =
mdtMetaClient.getActiveTimeline().getWriteTimeline().filterCompletedInstants().lastInstant().get();
+ final HoodieCommitMetadata mdtCommitMetadata =
mdtMetaClient.getTimelineLayout().getCommitMetadataSerDe().deserialize(
+ latestInstant,
+
mdtMetaClient.getActiveTimeline().getInstantDetails(latestInstant).get(),
+ HoodieCommitMetadata.class);
+ if
(mdtCommitMetadata.getPartitionToWriteStats().containsKey(MetadataPartitionType.COLUMN_STATS.getPartitionPath()))
{
+ // update data table's table config for list of columns indexed.
+ List<String> columnsToIndex =
HoodieTableMetadataUtil.getColumnsToIndex(commitMetadata,
dataTable.getMetaClient(), config.getMetadataConfig());
+ // if col stats is getting updated, lets also update list of
columns indexed if changed.
+ updateColSatsFunc.apply(dataTable.getMetaClient(), columnsToIndex);
+ }
+ } catch (Exception e) {
+ throw new HoodieException("Updating data table config to latest set
of columns indexed with col stats failed ", e);
+ }
+ }
+ }
+ }
+
+ /**
+ * Deletes col stats index definition for the given table of interest.
+ * @param dataTableMetaClient {@link HoodieTableMetaClient} instance for the
data table.
+ */
+ @VisibleForTesting
+ public static void deleteColumnStatsIndexDefinition(HoodieTableMetaClient
dataTableMetaClient) {
+ dataTableMetaClient.deleteIndexDefinition(PARTITION_NAME_COLUMN_STATS);
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
index a85a7497241..ebdf2916a78 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
@@ -76,6 +76,7 @@ import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.util.Lazy;
import org.apache.avro.Schema;
import org.slf4j.Logger;
@@ -98,6 +99,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static
org.apache.hudi.common.config.HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS;
import static
org.apache.hudi.common.config.HoodieMetadataConfig.DEFAULT_METADATA_POPULATE_META_FIELDS;
import static
org.apache.hudi.common.table.HoodieTableConfig.TIMELINE_HISTORY_PATH;
import static
org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED;
@@ -403,6 +405,7 @@ public abstract class HoodieBackedTableMetadataWriter<I>
implements HoodieTableM
LOG.info("Initializing MDT partition {} at instant {}",
partitionTypeName, instantTimeForPartition);
String partitionName;
Pair<Integer, HoodieData<HoodieRecord>> fileGroupCountAndRecordsPair;
+ List<String> columnsToIndex = new ArrayList<>();
try {
switch (partitionType) {
case FILES:
@@ -414,7 +417,9 @@ public abstract class HoodieBackedTableMetadataWriter<I>
implements HoodieTableM
partitionName = BLOOM_FILTERS.getPartitionPath();
break;
case COLUMN_STATS:
- fileGroupCountAndRecordsPair =
initializeColumnStatsPartition(partitionToFilesMap);
+ Pair<List<String>, Pair<Integer, HoodieData<HoodieRecord>>>
colStatsColumnsAndRecord = initializeColumnStatsPartition(partitionToFilesMap);
+ columnsToIndex = colStatsColumnsAndRecord.getKey();
+ fileGroupCountAndRecordsPair = colStatsColumnsAndRecord.getValue();
partitionName = COLUMN_STATS.getPartitionPath();
break;
case RECORD_INDEX:
@@ -477,7 +482,10 @@ public abstract class HoodieBackedTableMetadataWriter<I>
implements HoodieTableM
HoodieData<HoodieRecord> records =
fileGroupCountAndRecordsPair.getValue();
bulkCommit(instantTimeForPartition, partitionName, records,
fileGroupCount);
metadataMetaClient.reloadActiveTimeline();
-
+ if (partitionType == COLUMN_STATS) {
+ // initialize Col Stats index definition
+ updateColumnsToIndexWithColStats(columnsToIndex);
+ }
dataMetaClient.getTableConfig().setMetadataPartitionState(dataMetaClient,
partitionName, true);
// initialize the metadata reader again so the MDT partition can be read
after initialization
initMetadataReader();
@@ -486,6 +494,12 @@ public abstract class HoodieBackedTableMetadataWriter<I>
implements HoodieTableM
}
}
+ /**
+ * Updates the list of columns to index with col stats index.
+ * @param columnsToIndex list of columns to index.
+ */
+ protected abstract void updateColumnsToIndexWithColStats(List<String>
columnsToIndex);
+
/**
* Returns a unique timestamp to use for initializing a MDT partition.
* <p>
@@ -519,15 +533,28 @@ public abstract class HoodieBackedTableMetadataWriter<I>
implements HoodieTableM
return Pair.of(fileGroupCount, records);
}
- private Pair<Integer, HoodieData<HoodieRecord>>
initializeColumnStatsPartition(Map<String, Map<String, Long>>
partitionToFilesMap) {
+ private Pair<List<String>, Pair<Integer, HoodieData<HoodieRecord>>>
initializeColumnStatsPartition(Map<String, Map<String, Long>>
partitionToFilesMap) {
+ // Find the columns to index
+ final List<String> columnsToIndex =
HoodieTableMetadataUtil.getColumnsToIndex(dataMetaClient.getTableConfig(),
+ dataWriteConfig.getMetadataConfig(), Lazy.lazily(() ->
HoodieTableMetadataUtil.tryResolveSchemaForTable(dataMetaClient)));
+ if (columnsToIndex.isEmpty()) {
+ // atleast meta fields will be chosen for indexing. So, we should not
reach this state unless virtual keys are enabled and
+ // all col types are unsupported among the ones configured.
+ throw new HoodieException("No columns are found eligible to be indexed.
Can you try setting right value for " + COLUMN_STATS_INDEX_FOR_COLUMNS.key()
+ + ". Current value set : " +
dataWriteConfig.getMetadataConfig().getColumnsEnabledForColumnStatsIndex());
+ }
+
+ LOG.info("Indexing {} columns for column stats index",
columnsToIndex.size());
+
// during initialization, we need stats for base and log files.
HoodieData<HoodieRecord> records =
HoodieTableMetadataUtil.convertFilesToColumnStatsRecords(
engineContext, Collections.emptyMap(), partitionToFilesMap,
dataMetaClient, dataWriteConfig.getMetadataConfig(),
dataWriteConfig.getColumnStatsIndexParallelism(),
- dataWriteConfig.getMetadataConfig().getMaxReaderBufferSize());
+ dataWriteConfig.getMetadataConfig().getMaxReaderBufferSize(),
+ columnsToIndex);
final int fileGroupCount =
dataWriteConfig.getMetadataConfig().getColumnStatsIndexFileGroupCount();
- return Pair.of(fileGroupCount, records);
+ return Pair.of(columnsToIndex, Pair.of(fileGroupCount, records));
}
private Pair<Integer, HoodieData<HoodieRecord>>
initializeBloomFiltersPartition(String createInstantTime, Map<String,
Map<String, Long>> partitionToFilesMap) {
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
index 19b6624922a..1cc04b9860e 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
@@ -31,6 +31,7 @@ import org.apache.hudi.avro.model.HoodieRestorePlan;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.avro.model.HoodieRollbackPlan;
import org.apache.hudi.avro.model.HoodieSavepointMetadata;
+import org.apache.hudi.client.HoodieColumnStatsIndexUtils;
import org.apache.hudi.common.HoodiePendingRollbackInfo;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -1029,6 +1030,10 @@ public abstract class HoodieTable<T, I, K, O> implements
Serializable {
if (metadataPartitionExists(metaClient.getBasePath(), context,
partitionType.getPartitionPath())) {
deleteMetadataPartition(metaClient.getBasePath(), context,
partitionType.getPartitionPath());
}
+ if (partitionType == MetadataPartitionType.COLUMN_STATS) {
+ // delete index definition as well
+
HoodieColumnStatsIndexUtils.deleteColumnStatsIndexDefinition(getMetaClient());
+ }
clearMetadataTablePartitionsConfig(Option.of(partitionType), false);
} catch (HoodieMetadataException e) {
throw new HoodieException("Failed to delete metadata partition: " +
partitionType.name(), e);
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
index 841a3553dd0..7b51bf35ee0 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
@@ -21,6 +21,7 @@ package org.apache.hudi.table.action.commit;
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.avro.model.HoodieClusteringGroup;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.HoodieColumnStatsIndexUtils;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.transaction.TransactionManager;
import org.apache.hudi.client.utils.TransactionUtils;
@@ -33,12 +34,14 @@ import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieInstant.State;
import org.apache.hudi.common.table.timeline.InstantGenerator;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CommitUtils;
+import org.apache.hudi.common.util.Functions;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.StringUtils;
@@ -230,12 +233,25 @@ public abstract class BaseCommitActionExecutor<T, I, K,
O, R>
serializeCommitMetadata(table.getMetaClient().getCommitMetadataSerDe(),
metadata));
LOG.info("Committed " + instantTime);
result.setCommitMetadata(Option.of(metadata));
+ // update cols to Index as applicable
+ HoodieColumnStatsIndexUtils.updateColsToIndex(table, config, metadata,
+ (Functions.Function2<HoodieTableMetaClient, List<String>, Void>)
(metaClient, columnsToIndex) -> {
+ updateColumnsToIndexForColumnStats(metaClient, columnsToIndex);
+ return null;
+ });
} catch (IOException e) {
throw new HoodieCommitException("Failed to complete commit " +
config.getBasePath() + " at time " + instantTime,
e);
}
}
+ /**
+ * Updates the list of columns indexed with col stats index in Metadata
table.
+ * @param metaClient instance of {@link HoodieTableMetaClient} of interest.
+ * @param columnsToIndex list of columns to index.
+ */
+ protected abstract void
updateColumnsToIndexForColumnStats(HoodieTableMetaClient metaClient,
List<String> columnsToIndex);
+
/**
* Finalize Write operation.
* @param instantTime Instant Time
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/functional/BaseHoodieIndexClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/functional/BaseHoodieIndexClient.java
index 2fb4717d8a5..72c12d27d2a 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/functional/BaseHoodieIndexClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/functional/BaseHoodieIndexClient.java
@@ -28,6 +28,7 @@ import org.apache.hudi.storage.StoragePath;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
import java.util.Map;
public abstract class BaseHoodieIndexClient {
@@ -47,12 +48,15 @@ public abstract class BaseHoodieIndexClient {
public void register(HoodieTableMetaClient metaClient, HoodieIndexDefinition
indexDefinition) {
LOG.info("Registering index {} of using {}",
indexDefinition.getIndexName(), indexDefinition.getIndexType());
// build HoodieIndexMetadata and then add to index definition file
- metaClient.buildIndexDefinition(indexDefinition);
- // update table config if necessary
- String indexMetaPath = metaClient.getIndexDefinitionPath();
- if
(!metaClient.getTableConfig().getProps().containsKey(HoodieTableConfig.RELATIVE_INDEX_DEFINITION_PATH)
|| !metaClient.getTableConfig().getRelativeIndexDefinitionPath().isPresent()) {
-
metaClient.getTableConfig().setValue(HoodieTableConfig.RELATIVE_INDEX_DEFINITION_PATH,
FSUtils.getRelativePartitionPath(metaClient.getBasePath(), new
StoragePath(indexMetaPath)));
- HoodieTableConfig.update(metaClient.getStorage(),
metaClient.getMetaPath(), metaClient.getTableConfig().getProps());
+ boolean indexDefnUpdated =
metaClient.buildIndexDefinition(indexDefinition);
+ if (indexDefnUpdated) {
+ String indexMetaPath = metaClient.getIndexDefinitionPath();
+ // update table config if necessary
+ if
(!metaClient.getTableConfig().getProps().containsKey(HoodieTableConfig.RELATIVE_INDEX_DEFINITION_PATH.key())
+ ||
!metaClient.getTableConfig().getRelativeIndexDefinitionPath().isPresent()) {
+
metaClient.getTableConfig().setValue(HoodieTableConfig.RELATIVE_INDEX_DEFINITION_PATH,
FSUtils.getRelativePartitionPath(metaClient.getBasePath(), new
StoragePath(indexMetaPath)));
+ HoodieTableConfig.update(metaClient.getStorage(),
metaClient.getMetaPath(), metaClient.getTableConfig().getProps());
+ }
}
}
@@ -62,6 +66,13 @@ public abstract class BaseHoodieIndexClient {
public abstract void create(HoodieTableMetaClient metaClient, String
indexName, String indexType, Map<String, Map<String, String>> columns,
Map<String, String> options,
Map<String, String> tableProperties) throws
Exception;
+ /**
+ * Creates or updated the col stats index definition.
+ * @param metaClient data table's {@link HoodieTableMetaClient} instance.
+ * @param columnsToIndex list of columns to index.
+ */
+ public abstract void
createOrUpdateColumnStatsIndexDefinition(HoodieTableMetaClient metaClient,
List<String> columnsToIndex);
+
/**
* Drop an index. By default, ignore drop if index does not exist.
*
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
index ba3570a0b49..7eec5da54dd 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -430,6 +430,11 @@ public class HoodieFlinkWriteClient<T> extends
cleanHandles();
}
+ @Override
+ protected void updateColumnsToIndexWithColStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ //no op. HUDI-8801 to fix.
+ }
+
/**
* Get or create a new write handle in order to reuse the file handles.
*
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
index 5d63075c68c..b7d63623934 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
@@ -97,6 +97,11 @@ public class FlinkHoodieBackedTableMetadataWriter extends
HoodieBackedTableMetad
}
}
+ @Override
+ protected void updateColumnsToIndexWithColStats(List<String> columnsToIndex)
{
+ // no op. HUDI-8801 to fix.
+ }
+
@Override
protected void commit(String instantTime, Map<String,
HoodieData<HoodieRecord>> partitionRecordsMap) {
commitInternal(instantTime, partitionRecordsMap, false, Option.empty());
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
index 7a27830274f..e18052f002a 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
@@ -23,9 +23,11 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.execution.FlinkLazyInsertIterable;
import org.apache.hudi.io.ExplicitWriteHandleFactory;
@@ -197,4 +199,9 @@ public abstract class BaseFlinkCommitActionExecutor<T>
extends
return new FlinkLazyInsertIterable<>(recordItr, true, config, instantTime,
table, idPfx,
taskContextSupplier, new ExplicitWriteHandleFactory<>(writeHandle));
}
+
+ @Override
+ protected void updateColumnsToIndexForColumnStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ throw new HoodieNotSupportedException("Col stats with flink is not yet
supported");
+ }
}
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java
index ac463e69b0d..24819c8b697 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java
@@ -23,6 +23,7 @@ import org.apache.hudi.avro.model.HoodieClusteringPlan;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -43,6 +44,11 @@ public class HoodieJavaTableServiceClient<T> extends
BaseHoodieTableServiceClien
super(context, clientConfig, timelineService);
}
+ @Override
+ protected void updateColumnsToIndexWithColStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ // no op
+ }
+
@Override
protected void
validateClusteringCommit(HoodieWriteMetadata<List<WriteStatus>>
clusteringMetadata, String clusteringCommitTime, HoodieTable table) {
if (clusteringMetadata.getWriteStatuses().isEmpty()) {
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
index 4742c25c557..486b2d13557 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
@@ -57,6 +57,11 @@ public class HoodieJavaWriteClient<T> extends
this.tableServiceClient = new HoodieJavaTableServiceClient<>(context,
writeConfig, getTimelineServer());
}
+ @Override
+ protected void updateColumnsToIndexWithColStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ // no op
+ }
+
public HoodieJavaWriteClient(HoodieEngineContext context,
HoodieWriteConfig writeConfig,
boolean rollbackPending,
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieBackedTableMetadataWriter.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieBackedTableMetadataWriter.java
index 52bf184d2dd..c3e75a47694 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieBackedTableMetadataWriter.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieBackedTableMetadataWriter.java
@@ -92,6 +92,11 @@ public class JavaHoodieBackedTableMetadataWriter extends
HoodieBackedTableMetada
}
}
+ @Override
+ protected void updateColumnsToIndexWithColStats(List<String> columnsToIndex)
{
+ // no op. HUDI-8801 to fix.
+ }
+
@Override
protected void commit(String instantTime, Map<String,
HoodieData<HoodieRecord>> partitionRecordsMap) {
commitInternal(instantTime, partitionRecordsMap, false, Option.empty());
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
index f447f630d80..2a9c2b86024 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
@@ -298,4 +298,9 @@ public abstract class BaseJavaCommitActionExecutor<T>
extends
result.setPartitionToReplaceFileIds(getPartitionToReplacedFileIds(result));
commitOnAutoCommit(result);
}
+
+ @Override
+ protected void updateColumnsToIndexForColumnStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ // no op. HUDI-8801
+ }
}
diff --git
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/testutils/TestHoodieMetadataBase.java
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/testutils/TestHoodieMetadataBase.java
index 0e7f72cbd33..092f524f967 100644
---
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/testutils/TestHoodieMetadataBase.java
+++
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/testutils/TestHoodieMetadataBase.java
@@ -304,6 +304,7 @@ public class TestHoodieMetadataBase extends
HoodieJavaClientTestHarness {
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BLOOM).build())
.withMetadataConfig(HoodieMetadataConfig.newBuilder()
.enable(useFileListingMetadata)
+ .withMetadataIndexColumnStats(false)
.enableMetrics(enableMetrics)
.ignoreSpuriousDeletes(validateMetadataPayloadConsistency)
.build())
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/HoodieSparkIndexClient.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/HoodieSparkIndexClient.java
similarity index 70%
rename from
hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/HoodieSparkIndexClient.java
rename to
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/HoodieSparkIndexClient.java
index fd6f0ddd740..b318871a468 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/HoodieSparkIndexClient.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/HoodieSparkIndexClient.java
@@ -20,8 +20,13 @@
package org.apache.hudi;
import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.client.transaction.lock.FileSystemBasedLockProvider;
+import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.config.HoodieIndexingConfig;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.EngineType;
+import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieIndexDefinition;
import org.apache.hudi.common.model.HoodieIndexMetadata;
import org.apache.hudi.common.model.HoodieRecordPayload;
@@ -35,8 +40,10 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIndexException;
import org.apache.hudi.exception.HoodieMetadataIndexException;
import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.storage.StorageSchemes;
import org.apache.hudi.table.action.index.functional.BaseHoodieIndexClient;
+import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,8 +57,6 @@ import java.util.Map;
import scala.collection.JavaConverters;
-import static org.apache.hudi.HoodieConversionUtils.mapAsScalaImmutableMap;
-import static org.apache.hudi.HoodieConversionUtils.toScalaOption;
import static
org.apache.hudi.common.config.HoodieMetadataConfig.ENABLE_METADATA_INDEX_BLOOM_FILTER;
import static
org.apache.hudi.common.config.HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS;
import static
org.apache.hudi.common.config.HoodieMetadataConfig.RECORD_INDEX_ENABLE_PROP;
@@ -70,13 +75,23 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
private static final Logger LOG =
LoggerFactory.getLogger(HoodieSparkIndexClient.class);
- private static volatile HoodieSparkIndexClient _instance;
-
- private final SparkSession sparkSession;
+ private Option<SparkSession> sparkSessionOpt = Option.empty();
+ private Option<HoodieWriteConfig> writeConfigOpt = Option.empty();
+ private Option<HoodieEngineContext> engineContextOpt = Option.empty();
public HoodieSparkIndexClient(SparkSession sparkSession) {
+ this(Option.of(sparkSession), Option.empty(), Option.empty());
+ }
+
+ public HoodieSparkIndexClient(HoodieWriteConfig writeConfig,
HoodieEngineContext engineContext) {
+ this(Option.empty(), Option.of(writeConfig), Option.of(engineContext));
+ }
+
+ public HoodieSparkIndexClient(Option<SparkSession> sparkSessionOpt,
Option<HoodieWriteConfig> writeConfig, Option<HoodieEngineContext>
engineContext) {
super();
- this.sparkSession = sparkSession;
+ this.sparkSessionOpt = sparkSessionOpt;
+ this.writeConfigOpt = writeConfig;
+ this.engineContextOpt = engineContext;
}
@Override
@@ -101,8 +116,7 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
}
LOG.info("Creating index {} of using {}", fullIndexName, indexType);
- try (SparkRDDWriteClient writeClient =
HoodieCLIUtils.createHoodieWriteClient(sparkSession,
metaClient.getBasePath().toString(),
- mapAsScalaImmutableMap(buildWriteConfig(metaClient, Option.empty(),
Option.of(indexType))), toScalaOption(Option.empty()))) {
+ try (SparkRDDWriteClient writeClient = getWriteClient(metaClient,
Option.empty(), Option.of(indexType))) {
// generate index plan
Option<String> indexInstantTime = doSchedule(writeClient, metaClient,
fullIndexName, MetadataPartitionType.RECORD_INDEX);
if (indexInstantTime.isPresent()) {
@@ -117,6 +131,14 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
}
}
+ @Override
+ public void createOrUpdateColumnStatsIndexDefinition(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ HoodieIndexDefinition indexDefinition = new
HoodieIndexDefinition(PARTITION_NAME_COLUMN_STATS, PARTITION_NAME_COLUMN_STATS,
PARTITION_NAME_COLUMN_STATS,
+ columnsToIndex, Collections.EMPTY_MAP);
+ LOG.info("Registering Or Updating the index " +
PARTITION_NAME_COLUMN_STATS);
+ register(metaClient, indexDefinition);
+ }
+
private void createExpressionOrSecondaryIndex(HoodieTableMetaClient
metaClient, String userIndexName, String indexType,
Map<String, Map<String,
String>> columns, Map<String, String> options, Map<String, String>
tableProperties) throws Exception {
String fullIndexName = indexType.equals(PARTITION_NAME_SECONDARY_INDEX)
@@ -144,8 +166,7 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
LOG.info("Creating index {} of using {}", fullIndexName, indexType);
Option<HoodieIndexDefinition> expressionIndexDefinitionOpt =
Option.ofNullable(indexDefinition);
- try (SparkRDDWriteClient writeClient =
HoodieCLIUtils.createHoodieWriteClient(sparkSession,
metaClient.getBasePath().toString(),
- mapAsScalaImmutableMap(buildWriteConfig(metaClient,
expressionIndexDefinitionOpt, Option.of(indexType))),
toScalaOption(Option.empty()))) {
+ try (SparkRDDWriteClient writeClient = getWriteClient(metaClient,
expressionIndexDefinitionOpt, Option.of(indexType))) {
MetadataPartitionType partitionType =
indexType.equals(PARTITION_NAME_SECONDARY_INDEX) ?
MetadataPartitionType.SECONDARY_INDEX : MetadataPartitionType.EXPRESSION_INDEX;
// generate index plan
Option<String> indexInstantTime = doSchedule(writeClient, metaClient,
fullIndexName, partitionType);
@@ -163,8 +184,7 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
private void drop(HoodieTableMetaClient metaClient, String indexName,
Option<HoodieIndexDefinition> indexDefinitionOpt) {
LOG.info("Dropping index {}", indexName);
- try (SparkRDDWriteClient writeClient =
HoodieCLIUtils.createHoodieWriteClient(sparkSession,
metaClient.getBasePath().toString(),
- mapAsScalaImmutableMap(buildWriteConfig(metaClient,
indexDefinitionOpt, Option.empty())), toScalaOption(Option.empty()))) {
+ try (SparkRDDWriteClient writeClient = getWriteClient(metaClient,
indexDefinitionOpt, Option.empty())) {
writeClient.dropIndex(Collections.singletonList(indexName));
}
}
@@ -175,12 +195,50 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
Option<HoodieIndexDefinition> indexDefinitionOpt =
metaClient.getIndexMetadata()
.map(HoodieIndexMetadata::getIndexDefinitions)
.map(definition -> definition.get(indexName));
- try (SparkRDDWriteClient writeClient =
HoodieCLIUtils.createHoodieWriteClient(sparkSession,
metaClient.getBasePath().toString(),
- mapAsScalaImmutableMap(buildWriteConfig(metaClient,
indexDefinitionOpt, Option.empty())), toScalaOption(Option.empty()))) {
+ try (SparkRDDWriteClient writeClient = getWriteClient(metaClient,
indexDefinitionOpt, Option.empty())) {
writeClient.dropIndex(Collections.singletonList(indexName));
}
}
+ private SparkRDDWriteClient getWriteClient(HoodieTableMetaClient metaClient,
Option<HoodieIndexDefinition> indexDefinitionOpt,
+ Option<String> indexTypeOpt) {
+ try {
+ TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
+ String schemaStr = schemaUtil.getTableAvroSchema(false).toString();
+ TypedProperties props = getProps(metaClient, indexDefinitionOpt,
indexTypeOpt, schemaStr);
+ if (!engineContextOpt.isPresent()) {
+ engineContextOpt = Option.of(new HoodieSparkEngineContext(new
JavaSparkContext(sparkSessionOpt.get().sparkContext())));
+ }
+ HoodieWriteConfig localWriteConfig = HoodieWriteConfig.newBuilder()
+ .withPath(metaClient.getBasePath())
+ .withProperties(props)
+ .withEmbeddedTimelineServerEnabled(false)
+ .withAutoCommit(false)
+ .withSchema(schemaStr)
+ .withEngineType(EngineType.SPARK).build();
+ return new SparkRDDWriteClient(engineContextOpt.get(), localWriteConfig,
Option.empty());
+ } catch (Exception e) {
+ throw new HoodieException("Failed to create write client while
performing index operation ", e);
+ }
+ }
+
+ private TypedProperties getProps(HoodieTableMetaClient metaClient,
Option<HoodieIndexDefinition> indexDefinitionOpt,
+ Option<String> indexTypeOpt, String
schemaStr) {
+ if (writeConfigOpt.isPresent()) {
+ return writeConfigOpt.get().getProps();
+ } else {
+ TypedProperties typedProperties = metaClient.getTableConfig().getProps();
+
JavaConverters.mapAsJavaMapConverter(sparkSessionOpt.get().sqlContext().getAllConfs()).asJava().forEach((k,
v) -> {
+ if (k.startsWith("hoodie.")) {
+ typedProperties.put(k, v);
+ }
+ });
+ typedProperties.putAll(buildWriteConfig(metaClient, indexDefinitionOpt,
indexTypeOpt));
+ typedProperties.put(HoodieWriteConfig.AVRO_SCHEMA_STRING.key(),
schemaStr);
+ return typedProperties;
+ }
+ }
+
private static Option<String>
doSchedule(SparkRDDWriteClient<HoodieRecordPayload> client,
HoodieTableMetaClient metaClient, String indexName, MetadataPartitionType
partitionType) {
List<MetadataPartitionType> partitionTypes =
Collections.singletonList(partitionType);
if (metaClient.getTableConfig().getMetadataPartitions().isEmpty()) {
@@ -193,12 +251,13 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
return
metaClient.getTableConfig().getMetadataPartitions().stream().anyMatch(partition
-> partition.equals(indexName));
}
- private static Map<String, String> buildWriteConfig(HoodieTableMetaClient
metaClient, Option<HoodieIndexDefinition> indexDefinitionOpt, Option<String>
indexTypeOpt) {
+ private static Map<String, String> buildWriteConfig(HoodieTableMetaClient
metaClient, Option<HoodieIndexDefinition> indexDefinitionOpt,
+ Option<String>
indexTypeOpt) {
Map<String, String> writeConfig = new HashMap<>();
if (metaClient.getTableConfig().isMetadataTableAvailable()) {
writeConfig.put(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(),
WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.name());
-
writeConfig.putAll(JavaConverters.mapAsJavaMapConverter(HoodieCLIUtils.getLockOptions(metaClient.getBasePath().toString(),
- metaClient.getBasePath().toUri().getScheme(), new
TypedProperties())).asJava());
+ writeConfig.putAll(getLockOptions(metaClient.getBasePath().toString(),
+ metaClient.getBasePath().toUri().getScheme(), new
TypedProperties()));
// [HUDI-7472] Ensure write-config contains the existing MDT partition
to prevent those from getting deleted
metaClient.getTableConfig().getMetadataPartitions().forEach(partitionPath -> {
@@ -228,6 +287,18 @@ public class HoodieSparkIndexClient extends
BaseHoodieIndexClient {
return writeConfig;
}
+ static Map<String, String> getLockOptions(String tablePath, String scheme,
TypedProperties lockConfig) {
+ List<String> customSupportedFSs =
lockConfig.getStringList(HoodieCommonConfig.HOODIE_FS_ATOMIC_CREATION_SUPPORT.key(),
",", new ArrayList<String>());
+ if (scheme == null || customSupportedFSs.contains(scheme) ||
StorageSchemes.isAtomicCreationSupported(scheme)) {
+ TypedProperties props =
FileSystemBasedLockProvider.getLockConfig(tablePath);
+ Map<String, String> toReturn = new HashMap<>();
+ props.stringPropertyNames().stream().forEach(key -> toReturn.put(key,
props.getString(key)));
+ return toReturn;
+ } else {
+ return Collections.emptyMap();
+ }
+ }
+
private static boolean isEligibleForIndexing(HoodieTableMetaClient
metaClient, String indexType, Map<String, String> options, Map<String,
Map<String, String>> columns) throws Exception {
if (!validateDataTypeForSecondaryIndex(new ArrayList<>(columns.keySet()),
new TableSchemaResolver(metaClient).getTableAvroSchema())) {
return false;
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
index 23a2178fd03..090cafe6327 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -18,6 +18,7 @@
package org.apache.hudi.client;
+import org.apache.hudi.HoodieSparkIndexClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.client.utils.SparkReleaseResources;
@@ -337,6 +338,11 @@ public class SparkRDDWriteClient<T> extends
}
}
+ @Override
+ protected void updateColumnsToIndexWithColStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ new HoodieSparkIndexClient(config,
getEngineContext()).createOrUpdateColumnStatsIndexDefinition(metaClient,
columnsToIndex);
+ }
+
@Override
protected void releaseResources(String instantTime) {
SparkReleaseResources.releaseCachedData(context, config, basePath,
instantTime);
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
index c3fa8269bb7..4b9bf59fc6c 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
@@ -20,6 +20,7 @@ package org.apache.hudi.metadata;
import org.apache.hudi.AvroConversionUtils;
import org.apache.hudi.index.expression.HoodieSparkExpressionIndex;
+import org.apache.hudi.HoodieSparkIndexClient;
import org.apache.hudi.client.BaseHoodieWriteClient;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
@@ -252,4 +253,9 @@ public class SparkHoodieBackedTableMetadataWriter extends
HoodieBackedTableMetad
return HoodieJavaRDD.of(deletedRecords, sparkEngineContext, 1);
}
+
+ @Override
+ protected void updateColumnsToIndexWithColStats(List<String> columnsToIndex)
{
+ new HoodieSparkIndexClient(dataWriteConfig,
engineContext).createOrUpdateColumnStatsIndexDefinition(dataMetaClient,
columnsToIndex);
+ }
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
index 9b842064000..ce2b8e0650d 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
@@ -48,6 +48,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaRDD;
import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.keygen.KeyGeneratorInterface;
import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory;
import org.apache.hudi.storage.HoodieStorage;
@@ -219,6 +220,11 @@ public class SparkBootstrapCommitActionExecutor<T>
LOG.info("Committing metadata bootstrap !!");
}
+ @Override
+ protected void updateColumnsToIndexForColumnStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ throw new HoodieNotSupportedException("col stats is not supported with
bootstrap operation");
+ }
+
/**
* Perform Full Bootstrap.
* @param partitionFilesList List of partitions and files within that
partitions
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
index 94d8748166b..9e843a60681 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
@@ -18,6 +18,7 @@
package org.apache.hudi.table.action.commit;
+import org.apache.hudi.HoodieSparkIndexClient;
import org.apache.hudi.client.WriteStatus;
import
org.apache.hudi.client.clustering.update.strategy.SparkAllowUpdateStrategy;
import org.apache.hudi.client.utils.SparkValidatorUtils;
@@ -30,6 +31,7 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.HoodieTimer;
@@ -387,4 +389,9 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
protected void
runPrecommitValidators(HoodieWriteMetadata<HoodieData<WriteStatus>>
writeMetadata) {
SparkValidatorUtils.runValidators(config, writeMetadata, context, table,
instantTime);
}
+
+ @Override
+ protected void updateColumnsToIndexForColumnStats(HoodieTableMetaClient
metaClient, List<String> columnsToIndex) {
+ new HoodieSparkIndexClient(config,
context).createOrUpdateColumnStatsIndexDefinition(metaClient, columnsToIndex);
+ }
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieUnsafeUtils.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieUnsafeUtils.scala
index 393250b87d0..d698e42ecc1 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieUnsafeUtils.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieUnsafeUtils.scala
@@ -19,7 +19,6 @@
package org.apache.spark.sql
import org.apache.hudi.{HoodieUnsafeRDD, SparkAdapterSupport}
-
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
@@ -28,6 +27,8 @@ import org.apache.spark.sql.execution.LogicalRDD
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.MutablePair
+import scala.collection.JavaConverters
+
/**
* Suite of utilities helping in handling instances of [[HoodieUnsafeRDD]]
*/
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieSparkIndex.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieSparkIndex.java
new file mode 100644
index 00000000000..c1fa278f1a4
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieSparkIndex.java
@@ -0,0 +1,153 @@
+/*
+ * 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.hudi.client;
+
+import org.apache.hudi.HoodieSparkIndexClient;
+import org.apache.hudi.common.model.HoodieIndexDefinition;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.testutils.HoodieClientTestBase;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static
org.apache.hudi.index.expression.ExpressionIndexSparkFunctions.IDENTITY_FUNCTION;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_EXPRESSION_INDEX;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_EXPRESSION_INDEX_PREFIX;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX;
+import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestHoodieSparkIndex extends HoodieClientTestBase {
+
+ @Test
+ public void testIndexCreateAndDrop() throws IOException {
+ HoodieWriteConfig cfg = getConfigBuilder().build();
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(0x17AB);
+
+ initMetaClient(HoodieTableType.COPY_ON_WRITE);
+
+ try (SparkRDDWriteClient client = getHoodieWriteClient(cfg)) {
+ String commitTime1 = "001";
+ client.startCommitWithTime(commitTime1);
+ List<HoodieRecord> records1 = dataGen.generateInserts(commitTime1, 200);
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(records1, 1);
+ List<WriteStatus> statuses1 = client.upsert(writeRecords1,
commitTime1).collect();
+ assertNoWriteErrors(statuses1);
+ }
+
+ HoodieSparkIndexClient sparkIndexClient = new HoodieSparkIndexClient(cfg,
context);
+
+ // add 4 indices(2 SI and 2 EI)
+ String indexNamePrefix = "index";
+ String fieldNamePrefix = "field";
+ for (int i = 0; i < 5; i++) {
+ String indexName = indexNamePrefix + "_" + i;
+ String fieldName = fieldNamePrefix + "_" + i;
+ HoodieIndexDefinition indexDefinition = getIndexDefinition(indexName,
+ i % 2 == 0 ? PARTITION_NAME_SECONDARY_INDEX :
PARTITION_NAME_EXPRESSION_INDEX, fieldName);
+ sparkIndexClient.register(metaClient, indexDefinition);
+ readAndValidateIndexDefn(indexDefinition);
+ }
+
+ // add col stats index.
+ HoodieIndexDefinition colStatsIndexDefinition =
getIndexDefinition(PARTITION_NAME_COLUMN_STATS,
+ PARTITION_NAME_COLUMN_STATS, fieldNamePrefix + "_5");
+ sparkIndexClient.register(metaClient, colStatsIndexDefinition);
+ readAndValidateIndexDefn(colStatsIndexDefinition);
+
+ // drop one among sec index and expression index.
+ metaClient.deleteIndexDefinition(getIndexFullName(indexNamePrefix + "_1",
PARTITION_NAME_EXPRESSION_INDEX));
+ metaClient.deleteIndexDefinition(getIndexFullName(indexNamePrefix + "_2",
PARTITION_NAME_SECONDARY_INDEX));
+
+ //validate rest.
+ HoodieTableMetaClient newMetaClient =
HoodieTableMetaClient.builder().setBasePath(metaClient.getBasePath()).setConf(metaClient.getStorageConf())
+ .build();
+ HoodieIndexDefinition indexDefinition = getIndexDefinition(indexNamePrefix
+ "_0", PARTITION_NAME_SECONDARY_INDEX, fieldNamePrefix + "_0");
+ readAndValidateIndexDefn(indexDefinition, newMetaClient);
+ indexDefinition = getIndexDefinition(indexNamePrefix + "_3",
PARTITION_NAME_EXPRESSION_INDEX, fieldNamePrefix + "_3");
+ readAndValidateIndexDefn(indexDefinition, newMetaClient);
+ indexDefinition = getIndexDefinition(indexNamePrefix + "_4",
PARTITION_NAME_SECONDARY_INDEX, fieldNamePrefix + "_4");
+ readAndValidateIndexDefn(indexDefinition, newMetaClient);
+ readAndValidateIndexDefn(colStatsIndexDefinition);
+
+ // update col stats w/ new set of cols
+ List<String> colsToIndex = IntStream.range(0, 10).mapToObj(number ->
fieldNamePrefix + "_" + number).collect(Collectors.toList());
+ colStatsIndexDefinition = getIndexDefinition(PARTITION_NAME_COLUMN_STATS,
+ PARTITION_NAME_COLUMN_STATS, PARTITION_NAME_COLUMN_STATS, colsToIndex,
Collections.EMPTY_MAP);
+ sparkIndexClient.register(metaClient, colStatsIndexDefinition);
+ readAndValidateIndexDefn(colStatsIndexDefinition);
+
+ // drop col stats
+ metaClient.deleteIndexDefinition(colStatsIndexDefinition.getIndexName());
+ readAndValidateIndexDefnNotPresent(colStatsIndexDefinition,
HoodieTableMetaClient.builder().setBasePath(metaClient.getBasePath()).setConf(metaClient.getStorageConf())
+ .build());
+ }
+
+ private void readAndValidateIndexDefn(HoodieIndexDefinition
expectedIndexDefn) {
+ HoodieTableMetaClient newMetaClient =
HoodieTableMetaClient.builder().setBasePath(metaClient.getBasePath()).setConf(metaClient.getStorageConf())
+ .build();
+ readAndValidateIndexDefn(expectedIndexDefn, newMetaClient);
+ }
+
+ private void readAndValidateIndexDefn(HoodieIndexDefinition
expectedIndexDefn, HoodieTableMetaClient metaClient) {
+ assertTrue(metaClient.getIndexMetadata().isPresent());
+
assertTrue(!metaClient.getIndexMetadata().get().getIndexDefinitions().isEmpty());
+
assertTrue(metaClient.getIndexMetadata().get().getIndexDefinitions().containsKey(expectedIndexDefn.getIndexName()));
+ assertEquals(expectedIndexDefn,
metaClient.getIndexMetadata().get().getIndexDefinitions().get(expectedIndexDefn.getIndexName()));
+ }
+
+ private void readAndValidateIndexDefnNotPresent(HoodieIndexDefinition
expectedIndexDefn, HoodieTableMetaClient metaClient) {
+ assertTrue(metaClient.getIndexMetadata().isPresent());
+
assertTrue(!metaClient.getIndexMetadata().get().getIndexDefinitions().isEmpty());
+
assertTrue(!metaClient.getIndexMetadata().get().getIndexDefinitions().containsKey(expectedIndexDefn.getIndexName()));
+ }
+
+ private HoodieIndexDefinition getIndexDefinition(String indexName, String
indexType, String sourceField) {
+ return getIndexDefinition(indexName, indexType, IDENTITY_FUNCTION,
Collections.singletonList(sourceField), Collections.emptyMap());
+ }
+
+ private HoodieIndexDefinition getIndexDefinition(String indexName, String
indexType, String indexFunc, List<String> sourceFields,
+ Map<String, String>
indexOptions) {
+ String fullIndexName = getIndexFullName(indexName, indexType);
+ return new HoodieIndexDefinition(fullIndexName, indexType, indexFunc,
sourceFields, indexOptions);
+ }
+
+ private String getIndexFullName(String indexName, String indexType) {
+ if (indexName.equals(PARTITION_NAME_COLUMN_STATS)) {
+ return PARTITION_NAME_COLUMN_STATS;
+ }
+ return indexType.equals(PARTITION_NAME_SECONDARY_INDEX)
+ ? PARTITION_NAME_SECONDARY_INDEX_PREFIX + indexName
+ : PARTITION_NAME_EXPRESSION_INDEX_PREFIX + indexName;
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieIndexDefinition.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieIndexDefinition.java
index cbbbd730dba..2818a9f216e 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieIndexDefinition.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieIndexDefinition.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.StringJoiner;
import static org.apache.hudi.common.util.StringUtils.EMPTY_STRING;
@@ -135,4 +136,23 @@ public class HoodieIndexDefinition implements Serializable
{
.add("indexOptions=" + indexOptions)
.toString();
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof HoodieIndexDefinition)) {
+ return false;
+ }
+ HoodieIndexDefinition that = (HoodieIndexDefinition) o;
+ return getIndexName().equals(that.getIndexName()) &&
getIndexType().equals(that.getIndexType())
+ && getIndexFunction().equals(that.getIndexFunction()) &&
getSourceFields().equals(that.getSourceFields())
+ && getIndexOptions().equals(that.getIndexOptions());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(getIndexName(), getIndexType(), getIndexFunction(),
getSourceFields(), getIndexOptions());
+ }
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
index 8a49d012b71..991194cf61a 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
@@ -91,6 +91,7 @@ import static
org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
import static org.apache.hudi.common.util.ValidationUtils.checkState;
import static org.apache.hudi.io.storage.HoodieIOFactory.getIOFactory;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
/**
* <code>HoodieTableMetaClient</code> allows to access meta-data about a
hoodie table It returns meta-data about
@@ -211,25 +212,51 @@ public class HoodieTableMetaClient implements
Serializable {
/**
* Builds expression index definition and writes to index definition file.
+ * Support mutable and immutable index definition. Only col stats is
mutable, while all others are immutable.
+ * Inacse of immutable index definition, we could only create or delete the
definition.
+ * Incase of mutable (col stats), list of source columns (or list of columns
to index) could also change.
+ * @return true if index definition is updated.
*/
- public void buildIndexDefinition(HoodieIndexDefinition indexDefinition) {
+ public boolean buildIndexDefinition(HoodieIndexDefinition indexDefinition) {
String indexName = indexDefinition.getIndexName();
- checkState(
- !indexMetadataOpt.isPresent() ||
!indexMetadataOpt.get().getIndexDefinitions().containsKey(indexName),
- "Index metadata is already present");
+ boolean isIndexDefnImmutable =
!indexDefinition.getIndexName().equals(PARTITION_NAME_COLUMN_STATS); // only
col stats is mutable.
+ if (isIndexDefnImmutable) {
+ checkState(
+ !indexMetadataOpt.isPresent() ||
(!indexMetadataOpt.get().getIndexDefinitions().containsKey(indexName)),
+ "Index metadata is already present");
+ }
String indexMetaPath = getIndexDefinitionPath();
+ boolean updateIndexDefn = true;
if (indexMetadataOpt.isPresent()) {
- indexMetadataOpt.get().getIndexDefinitions().put(indexName,
indexDefinition);
+ if (isIndexDefnImmutable) {
+ indexMetadataOpt.get().getIndexDefinitions().put(indexName,
indexDefinition);
+ } else {
+ // if index defn is mutable, lets check for difference and only update
if required.
+ if
(indexMetadataOpt.get().getIndexDefinitions().containsKey(indexName)) {
+ if
(!indexMetadataOpt.get().getIndexDefinitions().get(indexName).getSourceFields().equals(indexDefinition.getSourceFields()))
{
+ LOG.info(String.format("List of columns to index is changing. Old
value %s. New value %s",
+
indexMetadataOpt.get().getIndexDefinitions().get(indexName).getSourceFields(),
indexDefinition.getSourceFields()));
+ indexMetadataOpt.get().getIndexDefinitions().put(indexName,
indexDefinition);
+ } else {
+ updateIndexDefn = false;
+ }
+ } else {
+ indexMetadataOpt.get().getIndexDefinitions().put(indexName,
indexDefinition);
+ }
+ }
} else {
Map<String, HoodieIndexDefinition> indexDefinitionMap = new HashMap<>();
indexDefinitionMap.put(indexName, indexDefinition);
indexMetadataOpt = Option.of(new
HoodieIndexMetadata(indexDefinitionMap));
}
- try {
- FileIOUtils.createFileInPath(storage, new StoragePath(indexMetaPath),
Option.of(getUTF8Bytes(indexMetadataOpt.get().toJson())));
- } catch (IOException e) {
- throw new HoodieIOException("Could not write expression index metadata
at path: " + indexMetaPath, e);
+ if (updateIndexDefn) {
+ try {
+ FileIOUtils.createFileInPath(storage, new StoragePath(indexMetaPath),
Option.of(getUTF8Bytes(indexMetadataOpt.get().toJson())));
+ } catch (IOException e) {
+ throw new HoodieIOException("Could not write expression index metadata
at path: " + indexMetaPath, e);
+ }
}
+ return updateIndexDefn;
}
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 77594396623..ef6d47d4c66 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -1135,16 +1135,8 @@ public class HoodieTableMetadataUtil {
HoodieTableMetaClient dataMetaClient,
HoodieMetadataConfig metadataConfig,
int
columnStatsIndexParallelism,
- int
maxReaderBufferSize) {
- // Find the columns to index
- final List<String> columnsToIndex =
getColumnsToIndex(dataMetaClient.getTableConfig(),
- metadataConfig, Lazy.lazily(() ->
tryResolveSchemaForTable(dataMetaClient)));
- if (columnsToIndex.isEmpty()) {
- // In case there are no columns to index, bail
- LOG.warn("No columns to index for column stats index.");
- return engineContext.emptyHoodieData();
- }
-
+ int
maxReaderBufferSize,
+
List<String> columnsToIndex) {
LOG.info("Indexing {} columns for column stats index",
columnsToIndex.size());
// Create the tuple (partition, filename, isDeleted) to handle both
deletes and appends
@@ -1335,21 +1327,7 @@ public class HoodieTableMetadataUtil {
}
try {
- Option<Schema> writerSchema =
-
Option.ofNullable(commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY))
- .flatMap(writerSchemaStr ->
- isNullOrEmpty(writerSchemaStr)
- ? Option.empty()
- : Option.of(new Schema.Parser().parse(writerSchemaStr)));
-
- HoodieTableConfig tableConfig = dataMetaClient.getTableConfig();
-
- // NOTE: Writer schema added to commit metadata will not contain Hudi's
metadata fields
- Option<Schema> tableSchema = writerSchema.map(schema ->
- tableConfig.populateMetaFields() ? addMetadataFields(schema) :
schema);
-
- List<String> columnsToIndex =
getColumnsToIndex(dataMetaClient.getTableConfig(), metadataConfig,
- Lazy.eagerly(tableSchema));
+ List<String> columnsToIndex = getColumnsToIndex(commitMetadata,
dataMetaClient, metadataConfig);
if (columnsToIndex.isEmpty()) {
// In case there are no columns to index, bail
return engineContext.emptyHoodieData();
@@ -1364,17 +1342,37 @@ public class HoodieTableMetadataUtil {
}
}
+ public static List<String> getColumnsToIndex(HoodieCommitMetadata
commitMetadata, HoodieTableMetaClient dataMetaClient, HoodieMetadataConfig
metadataConfig) {
+ Option<Schema> writerSchema =
+
Option.ofNullable(commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY))
+ .flatMap(writerSchemaStr ->
+ isNullOrEmpty(writerSchemaStr)
+ ? Option.empty()
+ : Option.of(new Schema.Parser().parse(writerSchemaStr)));
+
+ HoodieTableConfig tableConfig = dataMetaClient.getTableConfig();
+
+ // NOTE: Writer schema added to commit metadata will not contain Hudi's
metadata fields
+ Option<Schema> tableSchema = writerSchema.map(schema ->
+ tableConfig.populateMetaFields() ? addMetadataFields(schema) : schema);
+
+ return getColumnsToIndex(dataMetaClient.getTableConfig(), metadataConfig,
+ Lazy.eagerly(tableSchema));
+ }
+
@VisibleForTesting
static final String[] META_COLS_TO_ALWAYS_INDEX =
{COMMIT_TIME_METADATA_FIELD, RECORD_KEY_METADATA_FIELD,
PARTITION_PATH_METADATA_FIELD};
@VisibleForTesting
public static final Set<String> META_COL_SET_TO_INDEX = new
HashSet<>(Arrays.asList(META_COLS_TO_ALWAYS_INDEX));
+ @VisibleForTesting
public static List<String> getColumnsToIndex(HoodieTableConfig tableConfig,
HoodieMetadataConfig
metadataConfig,
List<String> columnNames) {
return getColumnsToIndex(tableConfig, metadataConfig,
Either.left(columnNames), Option.empty());
}
+ @VisibleForTesting
public static List<String> getColumnsToIndex(HoodieTableConfig tableConfig,
HoodieMetadataConfig
metadataConfig,
Lazy<Option<Schema>>
tableSchema,
@@ -1382,6 +1380,7 @@ public class HoodieTableMetadataUtil {
return getColumnsToIndex(tableConfig, metadataConfig,
Either.right(tableSchema), recordType);
}
+ @VisibleForTesting
public static List<String> getColumnsToIndex(HoodieTableConfig tableConfig,
HoodieMetadataConfig
metadataConfig,
Lazy<Option<Schema>>
tableSchema) {
@@ -1417,14 +1416,19 @@ public class HoodieTableMetadataUtil {
List<String> columnsToIndex =
metadataConfig.getColumnsEnabledForColumnStatsIndex();
if (!columnsToIndex.isEmpty()) {
// filter for top level fields here.
- List<String> topLevelFields = tableSchema.isLeft() ?
tableSchema.asLeft() :
- (tableSchema.asRight().get().map(schema ->
schema.getFields().stream().map(field ->
field.name()).collect(toList())).orElse(new ArrayList<String>()));
- return columnsToIndex.stream().filter(fieldName ->
!META_COL_SET_TO_INDEX.contains(fieldName) && (topLevelFields.isEmpty() ||
topLevelFields.contains(fieldName)));
+ List<String> topLevelEligibleFields = tableSchema.isLeft() ?
tableSchema.asLeft() :
+ (tableSchema.asRight().get().map(schema ->
schema.getFields().stream()
+ .filter(field -> isColumnTypeSupported(field.schema(),
recordType))
+ .map(field -> field.name()).collect(toList()))
+ .orElse(new ArrayList<String>()));
+ return columnsToIndex.stream().filter(fieldName ->
!META_COL_SET_TO_INDEX.contains(fieldName) && !fieldName.contains(".")
+ && (topLevelEligibleFields.isEmpty() ||
topLevelEligibleFields.contains(fieldName)));
}
if (tableSchema.isLeft()) {
return getFirstNFieldNames(tableSchema.asLeft().stream(),
metadataConfig.maxColumnsToIndexForColStats());
} else {
- return tableSchema.asRight().get().map(schema ->
getFirstNFieldNames(schema, metadataConfig.maxColumnsToIndexForColStats(),
recordType)).orElse(Stream.empty());
+ return tableSchema.asRight().get()
+ .map(schema -> getFirstNFieldNames(schema,
metadataConfig.maxColumnsToIndexForColStats(),
recordType)).orElse(Stream.empty());
}
}
@@ -1603,11 +1607,10 @@ public class HoodieTableMetadataUtil {
return value;
}
- private static Option<Schema> tryResolveSchemaForTable(HoodieTableMetaClient
dataTableMetaClient) {
+ public static Option<Schema> tryResolveSchemaForTable(HoodieTableMetaClient
dataTableMetaClient) {
if
(dataTableMetaClient.getCommitsTimeline().filterCompletedInstants().countInstants()
== 0) {
return Option.empty();
}
-
try {
TableSchemaResolver schemaResolver = new
TableSchemaResolver(dataTableMetaClient);
return Option.of(schemaResolver.getTableAvroSchema());
@@ -1683,11 +1686,6 @@ public class HoodieTableMetadataUtil {
}
private static boolean isColumnTypeSupported(Schema schema,
Option<HoodieRecordType> recordType) {
- // if record type is set and if its AVRO, MAP is unsupported.
- if (recordType.isPresent() && recordType.get() == HoodieRecordType.AVRO) {
- return schema.getType() != Schema.Type.MAP;
- }
- // if record Type is not set or if recordType is SPARK then we cannot
compare RECORD and ARRAY types in addition to MAP type
return schema.getType() != Schema.Type.RECORD && schema.getType() !=
Schema.Type.ARRAY && schema.getType() != Schema.Type.MAP;
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
b/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
index 662600f64e1..4089de3154b 100644
--- a/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
@@ -143,7 +143,7 @@ public class TestHoodieAvroUtils {
+ "{\"name\":\"decimal_col\",\"type\":[\"null\","
+
"{\"type\":\"bytes\",\"logicalType\":\"decimal\",\"precision\":8,\"scale\":4}],\"default\":null}]}";
- private static String SCHEMA_WITH_NESTED_FIELD =
"{\"name\":\"MyClass\",\"type\":\"record\",\"namespace\":\"com.acme.avro\",\"fields\":["
+ public static String SCHEMA_WITH_NESTED_FIELD =
"{\"name\":\"MyClass\",\"type\":\"record\",\"namespace\":\"com.acme.avro\",\"fields\":["
+ "{\"name\":\"firstname\",\"type\":\"string\"},"
+ "{\"name\":\"lastname\",\"type\":\"string\"},"
+
"{\"name\":\"student\",\"type\":{\"name\":\"student\",\"type\":\"record\",\"fields\":["
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
index a39465577c0..525e33eeb80 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
@@ -64,6 +64,7 @@ import java.util.UUID;
import java.util.stream.Collectors;
import static org.apache.hudi.avro.TestHoodieAvroUtils.SCHEMA_WITH_AVRO_TYPES;
+import static
org.apache.hudi.avro.TestHoodieAvroUtils.SCHEMA_WITH_NESTED_FIELD;
import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.getFileIDForFileGroup;
import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.validateDataTypeForPartitionStats;
import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.validateDataTypeForSecondaryIndex;
@@ -410,10 +411,10 @@ public class TestHoodieTableMetadataUtil extends
HoodieCommonTestHarness {
addNColumns(expected, tableSchemaSize);
assertEquals(expected,
HoodieTableMetadataUtil.getColumnsToIndex(tableConfig, metadataConfig,
colNames));
- //test with list of cols
+ //test with list of cols and a nested field as well.
metadataConfig = HoodieMetadataConfig.newBuilder()
.enable(true).withMetadataIndexColumnStats(true)
- .withColumnStatsIndexForColumns("col_1,col_7,col_11")
+ .withColumnStatsIndexForColumns("col_1,col_7,col_11,col12.col12_1")
.build();
colNames = new ArrayList<>();
addNColumns(colNames, 15);
@@ -462,7 +463,18 @@ public class TestHoodieTableMetadataUtil extends
HoodieCommonTestHarness {
expected.add("localTimestampMillisField");
assertEquals(expected,
HoodieTableMetadataUtil.getColumnsToIndex(tableConfig, metadataConfig,
Lazy.eagerly(Option.of(schema))));
+ //test with avro schema and nested fields and unsupported types
+ schema = new Schema.Parser().parse(SCHEMA_WITH_NESTED_FIELD);
+ metadataConfig = HoodieMetadataConfig.newBuilder()
+ .enable(true).withMetadataIndexColumnStats(true)
+ .withColumnStatsIndexForColumns("firstname,student.lastname,student")
+ .build();
+ expected = new
ArrayList<>(Arrays.asList(HoodieTableMetadataUtil.META_COLS_TO_ALWAYS_INDEX));
+ expected.add("firstname");
+ assertEquals(expected,
HoodieTableMetadataUtil.getColumnsToIndex(tableConfig, metadataConfig,
Lazy.eagerly(Option.of(schema))));
+
//test with avro schema with max cols set
+ schema = new Schema.Parser().parse(SCHEMA_WITH_AVRO_TYPES);
metadataConfig = HoodieMetadataConfig.newBuilder()
.enable(true).withMetadataIndexColumnStats(true)
.withMaxColumnsToIndexForColStats(2)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/cow-updated3-column-stats-index-table.json
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/cow-updated3-column-stats-index-table.json
new file mode 100644
index 00000000000..6a9f18efc63
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/cow-updated3-column-stats-index-table.json
@@ -0,0 +1,14 @@
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":1.329,"c3_minValue":1.329,"c3_nullCount":0,"c5_maxValue":310,"c5_minValue":310,"c5_nullCount":0,"c6_maxValue":"2020-03-28","c6_minValue":"2020-03-28","c6_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":10.329,"c3_minValue":10.329,"c3_nullCount":0,"c5_maxValue":99,"c5_minValue":99,"c5_nullCount":0,"c6_maxValue":"2020-03-28","c6_minValue":"2020-03-28","c6_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":568,"c1_minValue":8,"c1_nullCount":0,"c2_maxValue":"
8sdc","c2_minValue":"
111sdc","c2_nullCount":0,"c3_maxValue":979.272,"c3_minValue":82.111,"c3_nullCount":0,"c5_maxValue":58,"c5_minValue":2,"c5_nullCount":0,"c6_maxValue":"2020-11-08","c6_minValue":"2020-01-01","c6_nullCount":0,"c7_maxValue":"9g==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":15}
+{"c1_maxValue":715,"c1_minValue":76,"c1_nullCount":0,"c2_maxValue":"
76sdc","c2_minValue":"
224sdc","c2_nullCount":0,"c3_maxValue":958.579,"c3_minValue":246.427,"c3_nullCount":0,"c5_maxValue":73,"c5_minValue":9,"c5_nullCount":0,"c6_maxValue":"2020-11-21","c6_minValue":"2020-01-16","c6_nullCount":0,"c7_maxValue":"+g==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":12}
+{"c1_maxValue":768,"c1_minValue":59,"c1_nullCount":0,"c2_maxValue":"
768sdc","c2_minValue":"
118sdc","c2_nullCount":0,"c3_maxValue":959.131,"c3_minValue":64.768,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":7,"c5_nullCount":0,"c6_maxValue":"2020-11-20","c6_minValue":"2020-05-04","c6_nullCount":0,"c7_maxValue":"zw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":7}
+{"c1_maxValue":769,"c1_minValue":309,"c1_nullCount":0,"c2_maxValue":"
769sdc","c2_minValue":"
309sdc","c2_nullCount":0,"c3_maxValue":919.769,"c3_minValue":76.430,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":32,"c5_nullCount":0,"c6_maxValue":"2020-11-14","c6_minValue":"2020-01-08","c6_nullCount":0,"c7_maxValue":"uQ==","c7_minValue":"AQ==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":9}
+{"c1_maxValue":769,"c1_minValue":309,"c1_nullCount":0,"c2_maxValue":"
985sdc","c2_minValue":"
309sdc","c2_nullCount":0,"c3_maxValue":919.769,"c3_minValue":76.430,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":32,"c5_nullCount":0,"c6_maxValue":"2020-11-14","c6_minValue":"2020-01-08","c6_nullCount":0,"c7_maxValue":"uQ==","c7_minValue":"AQ==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":9}
+{"c1_maxValue":770,"c1_minValue":129,"c1_nullCount":0,"c2_maxValue":"
770sdc","c2_minValue":"
129sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":153.431,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":14,"c5_nullCount":0,"c6_maxValue":"2020-10-21","c6_minValue":"2020-01-15","c6_nullCount":0,"c7_maxValue":"rw==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":6}
+{"c1_maxValue":932,"c1_minValue":0,"c1_nullCount":0,"c2_maxValue":"
932sdc","c2_minValue":"
0sdc","c2_nullCount":0,"c3_maxValue":994.355,"c3_minValue":19.000,"c3_nullCount":0,"c5_maxValue":94,"c5_minValue":1,"c5_nullCount":0,"c6_maxValue":"2020-09-09","c6_minValue":"2020-01-01","c6_nullCount":0,"c7_maxValue":"xw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":8}
+{"c1_maxValue":932,"c1_minValue":0,"c1_nullCount":0,"c2_maxValue":"
987sdc","c2_minValue":"
0sdc","c2_nullCount":0,"c3_maxValue":994.355,"c3_minValue":19.000,"c3_nullCount":0,"c5_maxValue":94,"c5_minValue":1,"c5_nullCount":0,"c6_maxValue":"2020-10-10","c6_minValue":"2020-01-01","c6_nullCount":0,"c7_maxValue":"xw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":8}
+{"c1_maxValue":943,"c1_minValue":89,"c1_nullCount":0,"c2_maxValue":"
943sdc","c2_minValue":"
200sdc","c2_nullCount":0,"c3_maxValue":854.690,"c3_minValue":100.556,"c3_nullCount":0,"c5_maxValue":95,"c5_minValue":10,"c5_nullCount":0,"c6_maxValue":"2020-10-10","c6_minValue":"2020-01-10","c6_nullCount":0,"c7_maxValue":"yA==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":10}
+{"c1_maxValue":943,"c1_minValue":89,"c1_nullCount":0,"c2_maxValue":"
984sdc","c2_minValue":"
200sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":64.768,"c3_nullCount":1,"c5_maxValue":95,"c5_minValue":10,"c5_nullCount":0,"c6_maxValue":"2020-10-21","c6_minValue":"2020-01-10","c6_nullCount":0,"c7_maxValue":"yA==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":10}
+{"c1_maxValue":959,"c1_minValue":74,"c1_nullCount":0,"c2_maxValue":"
959sdc","c2_minValue":"
181sdc","c2_nullCount":0,"c3_maxValue":980.213,"c3_minValue":38.740,"c3_nullCount":0,"c5_maxValue":97,"c5_minValue":9,"c5_nullCount":0,"c6_maxValue":"2020-11-22","c6_minValue":"2020-01-23","c6_nullCount":0,"c7_maxValue":"1Q==","c7_minValue":"Kw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":13}
+{"c1_maxValue":959,"c1_minValue":74,"c1_nullCount":0,"c2_maxValue":"
989sdc","c2_minValue":"
181sdc","c2_nullCount":0,"c3_maxValue":980.213,"c3_minValue":38.740,"c3_nullCount":0,"c5_maxValue":97,"c5_minValue":9,"c5_nullCount":0,"c6_maxValue":"2020-11-22","c6_minValue":"2020-02-25","c6_nullCount":0,"c7_maxValue":"1Q==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":13}
\ No newline at end of file
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/cow-updated4-column-stats-index-table.json
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/cow-updated4-column-stats-index-table.json
new file mode 100644
index 00000000000..fbab3f86c78
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/cow-updated4-column-stats-index-table.json
@@ -0,0 +1,15 @@
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":1.129,"c3_minValue":1.129,"c3_nullCount":0,"c5_maxValue":10001,"c5_minValue":10001,"c5_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":1.329,"c3_minValue":1.329,"c3_nullCount":0,"c5_maxValue":310,"c5_minValue":310,"c5_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":10.329,"c3_minValue":10.329,"c3_nullCount":0,"c5_maxValue":99,"c5_minValue":99,"c5_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":568,"c1_minValue":8,"c1_nullCount":0,"c2_maxValue":"
8sdc","c2_minValue":"
111sdc","c2_nullCount":0,"c3_maxValue":979.272,"c3_minValue":82.111,"c3_nullCount":0,"c5_maxValue":58,"c5_minValue":2,"c5_nullCount":0,"c7_maxValue":"9g==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":15}
+{"c1_maxValue":715,"c1_minValue":76,"c1_nullCount":0,"c2_maxValue":"
76sdc","c2_minValue":"
224sdc","c2_nullCount":0,"c3_maxValue":958.579,"c3_minValue":246.427,"c3_nullCount":0,"c5_maxValue":73,"c5_minValue":9,"c5_nullCount":0,"c7_maxValue":"+g==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":12}
+{"c1_maxValue":768,"c1_minValue":59,"c1_nullCount":0,"c2_maxValue":"
768sdc","c2_minValue":"
118sdc","c2_nullCount":0,"c3_maxValue":959.131,"c3_minValue":64.768,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":7,"c5_nullCount":0,"c7_maxValue":"zw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":7}
+{"c1_maxValue":769,"c1_minValue":309,"c1_nullCount":0,"c2_maxValue":"
769sdc","c2_minValue":"
309sdc","c2_nullCount":0,"c3_maxValue":919.769,"c3_minValue":76.430,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":32,"c5_nullCount":0,"c7_maxValue":"uQ==","c7_minValue":"AQ==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":9}
+{"c1_maxValue":769,"c1_minValue":309,"c1_nullCount":0,"c2_maxValue":"
985sdc","c2_minValue":"
309sdc","c2_nullCount":0,"c3_maxValue":919.769,"c3_minValue":76.430,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":32,"c5_nullCount":0,"c7_maxValue":"uQ==","c7_minValue":"AQ==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":9}
+{"c1_maxValue":770,"c1_minValue":129,"c1_nullCount":0,"c2_maxValue":"
770sdc","c2_minValue":"
129sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":153.431,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":14,"c5_nullCount":0,"c7_maxValue":"rw==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":6}
+{"c1_maxValue":932,"c1_minValue":0,"c1_nullCount":0,"c2_maxValue":"
932sdc","c2_minValue":"
0sdc","c2_nullCount":0,"c3_maxValue":994.355,"c3_minValue":19.000,"c3_nullCount":0,"c5_maxValue":94,"c5_minValue":1,"c5_nullCount":0,"c7_maxValue":"xw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":8}
+{"c1_maxValue":932,"c1_minValue":0,"c1_nullCount":0,"c2_maxValue":"
987sdc","c2_minValue":"
0sdc","c2_nullCount":0,"c3_maxValue":994.355,"c3_minValue":19.000,"c3_nullCount":0,"c5_maxValue":94,"c5_minValue":1,"c5_nullCount":0,"c7_maxValue":"xw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":8}
+{"c1_maxValue":943,"c1_minValue":89,"c1_nullCount":0,"c2_maxValue":"
943sdc","c2_minValue":"
200sdc","c2_nullCount":0,"c3_maxValue":854.690,"c3_minValue":100.556,"c3_nullCount":0,"c5_maxValue":95,"c5_minValue":10,"c5_nullCount":0,"c7_maxValue":"yA==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":10}
+{"c1_maxValue":943,"c1_minValue":89,"c1_nullCount":0,"c2_maxValue":"
984sdc","c2_minValue":"
200sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":64.768,"c3_nullCount":1,"c5_maxValue":95,"c5_minValue":10,"c5_nullCount":0,"c7_maxValue":"yA==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":10}
+{"c1_maxValue":959,"c1_minValue":74,"c1_nullCount":0,"c2_maxValue":"
959sdc","c2_minValue":"
181sdc","c2_nullCount":0,"c3_maxValue":980.213,"c3_minValue":38.740,"c3_nullCount":0,"c5_maxValue":97,"c5_minValue":9,"c5_nullCount":0,"c7_maxValue":"1Q==","c7_minValue":"Kw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":13}
+{"c1_maxValue":959,"c1_minValue":74,"c1_nullCount":0,"c2_maxValue":"
989sdc","c2_minValue":"
181sdc","c2_nullCount":0,"c3_maxValue":980.213,"c3_minValue":38.740,"c3_nullCount":0,"c5_maxValue":97,"c5_minValue":9,"c5_nullCount":0,"c7_maxValue":"1Q==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":13}
\ No newline at end of file
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/mor-updated3-column-stats-index-table.json
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/mor-updated3-column-stats-index-table.json
new file mode 100644
index 00000000000..4840b9f0328
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/mor-updated3-column-stats-index-table.json
@@ -0,0 +1,14 @@
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":1.329,"c3_minValue":1.329,"c3_nullCount":0,"c5_maxValue":310,"c5_minValue":310,"c5_nullCount":0,"c6_maxValue":"2020-03-28","c6_minValue":"2020-03-28","c6_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":10.329,"c3_minValue":10.329,"c3_nullCount":0,"c5_maxValue":99,"c5_minValue":99,"c5_nullCount":0,"c6_maxValue":"2020-03-28","c6_minValue":"2020-03-28","c6_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":562,"c1_minValue":323,"c1_nullCount":0,"c2_maxValue":"
984sdc","c2_minValue":"
980sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":64.768,"c3_nullCount":1,"c5_maxValue":78,"c5_minValue":34,"c5_nullCount":0,"c6_maxValue":"2020-10-21","c6_minValue":"2020-01-15","c6_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"qw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":4}
+{"c1_maxValue":568,"c1_minValue":8,"c1_nullCount":0,"c2_maxValue":"
8sdc","c2_minValue":"
111sdc","c2_nullCount":0,"c3_maxValue":979.272,"c3_minValue":82.111,"c3_nullCount":0,"c5_maxValue":58,"c5_minValue":2,"c5_nullCount":0,"c6_maxValue":"2020-11-08","c6_minValue":"2020-01-01","c6_nullCount":0,"c7_maxValue":"9g==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":15}
+{"c1_maxValue":619,"c1_minValue":619,"c1_nullCount":0,"c2_maxValue":"
985sdc","c2_minValue":"
985sdc","c2_nullCount":0,"c3_maxValue":230.320,"c3_minValue":230.320,"c3_nullCount":0,"c5_maxValue":33,"c5_minValue":33,"c5_nullCount":0,"c6_maxValue":"2020-02-13","c6_minValue":"2020-02-13","c6_nullCount":0,"c7_maxValue":"QA==","c7_minValue":"QA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":633,"c1_minValue":624,"c1_nullCount":0,"c2_maxValue":"
987sdc","c2_minValue":"
986sdc","c2_nullCount":0,"c3_maxValue":580.317,"c3_minValue":375.308,"c3_nullCount":0,"c5_maxValue":33,"c5_minValue":32,"c5_nullCount":0,"c6_maxValue":"2020-10-10","c6_minValue":"2020-01-01","c6_nullCount":0,"c7_maxValue":"PQ==","c7_minValue":"NA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":2}
+{"c1_maxValue":639,"c1_minValue":555,"c1_nullCount":0,"c2_maxValue":"
989sdc","c2_minValue":"
982sdc","c2_nullCount":0,"c3_maxValue":904.304,"c3_minValue":153.431,"c3_nullCount":0,"c5_maxValue":44,"c5_minValue":31,"c5_nullCount":0,"c6_maxValue":"2020-08-25","c6_minValue":"2020-03-12","c6_nullCount":0,"c7_maxValue":"MA==","c7_minValue":"rw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":3}
+{"c1_maxValue":715,"c1_minValue":76,"c1_nullCount":0,"c2_maxValue":"
76sdc","c2_minValue":"
224sdc","c2_nullCount":0,"c3_maxValue":958.579,"c3_minValue":246.427,"c3_nullCount":0,"c5_maxValue":73,"c5_minValue":9,"c5_nullCount":0,"c6_maxValue":"2020-11-21","c6_minValue":"2020-01-16","c6_nullCount":0,"c7_maxValue":"+g==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":12}
+{"c1_maxValue":768,"c1_minValue":59,"c1_nullCount":0,"c2_maxValue":"
768sdc","c2_minValue":"
118sdc","c2_nullCount":0,"c3_maxValue":959.131,"c3_minValue":64.768,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":7,"c5_nullCount":0,"c6_maxValue":"2020-11-20","c6_minValue":"2020-05-04","c6_nullCount":0,"c7_maxValue":"zw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":7}
+{"c1_maxValue":769,"c1_minValue":309,"c1_nullCount":0,"c2_maxValue":"
769sdc","c2_minValue":"
309sdc","c2_nullCount":0,"c3_maxValue":919.769,"c3_minValue":76.430,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":32,"c5_nullCount":0,"c6_maxValue":"2020-11-14","c6_minValue":"2020-01-08","c6_nullCount":0,"c7_maxValue":"uQ==","c7_minValue":"AQ==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":9}
+{"c1_maxValue":770,"c1_minValue":129,"c1_nullCount":0,"c2_maxValue":"
770sdc","c2_minValue":"
129sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":153.431,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":14,"c5_nullCount":0,"c6_maxValue":"2020-10-21","c6_minValue":"2020-01-15","c6_nullCount":0,"c7_maxValue":"rw==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":6}
+{"c1_maxValue":932,"c1_minValue":0,"c1_nullCount":0,"c2_maxValue":"
932sdc","c2_minValue":"
0sdc","c2_nullCount":0,"c3_maxValue":994.355,"c3_minValue":19.000,"c3_nullCount":0,"c5_maxValue":94,"c5_minValue":1,"c5_nullCount":0,"c6_maxValue":"2020-09-09","c6_minValue":"2020-01-01","c6_nullCount":0,"c7_maxValue":"xw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":8}
+{"c1_maxValue":943,"c1_minValue":89,"c1_nullCount":0,"c2_maxValue":"
943sdc","c2_minValue":"
200sdc","c2_nullCount":0,"c3_maxValue":854.690,"c3_minValue":100.556,"c3_nullCount":0,"c5_maxValue":95,"c5_minValue":10,"c5_nullCount":0,"c6_maxValue":"2020-10-10","c6_minValue":"2020-01-10","c6_nullCount":0,"c7_maxValue":"yA==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":10}
+{"c1_maxValue":959,"c1_minValue":74,"c1_nullCount":0,"c2_maxValue":"
959sdc","c2_minValue":"
181sdc","c2_nullCount":0,"c3_maxValue":980.213,"c3_minValue":38.740,"c3_nullCount":0,"c5_maxValue":97,"c5_minValue":9,"c5_nullCount":0,"c6_maxValue":"2020-11-22","c6_minValue":"2020-01-23","c6_nullCount":0,"c7_maxValue":"1Q==","c7_minValue":"Kw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":13}
\ No newline at end of file
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/mor-updated4-column-stats-index-table.json
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/mor-updated4-column-stats-index-table.json
new file mode 100644
index 00000000000..9eba05658f8
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/mor-updated4-column-stats-index-table.json
@@ -0,0 +1,15 @@
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":1.129,"c3_minValue":1.129,"c3_nullCount":0,"c5_maxValue":10001,"c5_minValue":10001,"c5_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":1.329,"c3_minValue":1.329,"c3_nullCount":0,"c5_maxValue":310,"c5_minValue":310,"c5_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":101,"c1_minValue":101,"c1_nullCount":0,"c2_maxValue":"
999sdc","c2_minValue":"
999sdc","c2_nullCount":0,"c3_maxValue":10.329,"c3_minValue":10.329,"c3_nullCount":0,"c5_maxValue":99,"c5_minValue":99,"c5_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"SA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":562,"c1_minValue":323,"c1_nullCount":0,"c2_maxValue":"
984sdc","c2_minValue":"
980sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":64.768,"c3_nullCount":1,"c5_maxValue":78,"c5_minValue":34,"c5_nullCount":0,"c7_maxValue":"SA==","c7_minValue":"qw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":4}
+{"c1_maxValue":568,"c1_minValue":8,"c1_nullCount":0,"c2_maxValue":"
8sdc","c2_minValue":"
111sdc","c2_nullCount":0,"c3_maxValue":979.272,"c3_minValue":82.111,"c3_nullCount":0,"c5_maxValue":58,"c5_minValue":2,"c5_nullCount":0,"c7_maxValue":"9g==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":15}
+{"c1_maxValue":619,"c1_minValue":619,"c1_nullCount":0,"c2_maxValue":"
985sdc","c2_minValue":"
985sdc","c2_nullCount":0,"c3_maxValue":230.320,"c3_minValue":230.320,"c3_nullCount":0,"c5_maxValue":33,"c5_minValue":33,"c5_nullCount":0,"c7_maxValue":"QA==","c7_minValue":"QA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":1}
+{"c1_maxValue":633,"c1_minValue":624,"c1_nullCount":0,"c2_maxValue":"
987sdc","c2_minValue":"
986sdc","c2_nullCount":0,"c3_maxValue":580.317,"c3_minValue":375.308,"c3_nullCount":0,"c5_maxValue":33,"c5_minValue":32,"c5_nullCount":0,"c7_maxValue":"PQ==","c7_minValue":"NA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":2}
+{"c1_maxValue":639,"c1_minValue":555,"c1_nullCount":0,"c2_maxValue":"
989sdc","c2_minValue":"
982sdc","c2_nullCount":0,"c3_maxValue":904.304,"c3_minValue":153.431,"c3_nullCount":0,"c5_maxValue":44,"c5_minValue":31,"c5_nullCount":0,"c7_maxValue":"MA==","c7_minValue":"rw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":3}
+{"c1_maxValue":715,"c1_minValue":76,"c1_nullCount":0,"c2_maxValue":"
76sdc","c2_minValue":"
224sdc","c2_nullCount":0,"c3_maxValue":958.579,"c3_minValue":246.427,"c3_nullCount":0,"c5_maxValue":73,"c5_minValue":9,"c5_nullCount":0,"c7_maxValue":"+g==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":12}
+{"c1_maxValue":768,"c1_minValue":59,"c1_nullCount":0,"c2_maxValue":"
768sdc","c2_minValue":"
118sdc","c2_nullCount":0,"c3_maxValue":959.131,"c3_minValue":64.768,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":7,"c5_nullCount":0,"c7_maxValue":"zw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":7}
+{"c1_maxValue":769,"c1_minValue":309,"c1_nullCount":0,"c2_maxValue":"
769sdc","c2_minValue":"
309sdc","c2_nullCount":0,"c3_maxValue":919.769,"c3_minValue":76.430,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":32,"c5_nullCount":0,"c7_maxValue":"uQ==","c7_minValue":"AQ==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":9}
+{"c1_maxValue":770,"c1_minValue":129,"c1_nullCount":0,"c2_maxValue":"
770sdc","c2_minValue":"
129sdc","c2_nullCount":0,"c3_maxValue":977.328,"c3_minValue":153.431,"c3_nullCount":0,"c5_maxValue":78,"c5_minValue":14,"c5_nullCount":0,"c7_maxValue":"rw==","c7_minValue":"Ag==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":6}
+{"c1_maxValue":932,"c1_minValue":0,"c1_nullCount":0,"c2_maxValue":"
932sdc","c2_minValue":"
0sdc","c2_nullCount":0,"c3_maxValue":994.355,"c3_minValue":19.000,"c3_nullCount":0,"c5_maxValue":94,"c5_minValue":1,"c5_nullCount":0,"c7_maxValue":"xw==","c7_minValue":"AA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":8}
+{"c1_maxValue":943,"c1_minValue":89,"c1_nullCount":0,"c2_maxValue":"
943sdc","c2_minValue":"
200sdc","c2_nullCount":0,"c3_maxValue":854.690,"c3_minValue":100.556,"c3_nullCount":0,"c5_maxValue":95,"c5_minValue":10,"c5_nullCount":0,"c7_maxValue":"yA==","c7_minValue":"LA==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":10}
+{"c1_maxValue":959,"c1_minValue":74,"c1_nullCount":0,"c2_maxValue":"
959sdc","c2_minValue":"
181sdc","c2_nullCount":0,"c3_maxValue":980.213,"c3_minValue":38.740,"c3_nullCount":0,"c5_maxValue":97,"c5_minValue":9,"c5_nullCount":0,"c7_maxValue":"1Q==","c7_minValue":"Kw==","c7_nullCount":0,"c8_maxValue":9,"c8_minValue":9,"c8_nullCount":0,"valueCount":13}
\ No newline at end of file
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/update5-input-table-json/part-00000-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/update5-input-table-json/part-00000-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json
new file mode 100644
index 00000000000..e75a99130c1
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/update5-input-table-json/part-00000-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json
@@ -0,0 +1 @@
+{"c1":101,"c2":"
999sdc","c3":1.329,"c4":"2021-11-18T23:34:44.179-08:00","c5":310,"c6":"2020-03-28","c7":"SF==","c8":9}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/update6-input-table-json/part-00000-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/update6-input-table-json/part-00000-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json
new file mode 100644
index 00000000000..39b0c5c548e
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/resources/index/colstats/update6-input-table-json/part-00000-7e680484-e7e1-48b6-8289-1a7c483b530b-c000.json
@@ -0,0 +1 @@
+{"c1":101,"c2":"
999sdc","c3":1.129,"c4":"2021-11-18T23:34:44.179-08:00","c5":10001,"c6":"2020-03-28","c7":"SF==","c8":9}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
index 15bbd5263b6..807dd3e4d9e 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
@@ -33,20 +33,19 @@ import org.apache.hudi.storage.StoragePath
import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
import org.apache.hudi.testutils.HoodieSparkClientTestBase
import org.apache.hudi.{ColumnStatsIndexSupport, DataSourceWriteOptions}
-
import org.apache.spark.sql._
import org.apache.hudi.functional.ColumnStatIndexTestBase.ColumnStatsTestParams
import org.apache.hudi.metadata.HoodieTableMetadataUtil
+import
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS
import org.apache.hudi.testutils.{HoodieSparkClientTestBase,
LogFileColStatsTestUtil}
import org.apache.hudi.util.JavaScalaConverters.convertScalaListToJavaList
import org.apache.hudi.{ColumnStatsIndexSupport, DataSourceWriteOptions}
-
import org.apache.spark.sql.functions.typedLit
import org.apache.spark.sql.functions.{lit, struct, typedLit}
import org.apache.spark.sql.types._
import org.apache.spark.sql.DataFrame
import
org.apache.spark.sql.catalyst.dsl.expressions.StringToAttributeConversionHelper
-import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.api._
import org.junit.jupiter.params.provider.Arguments
@@ -55,7 +54,6 @@ import java.sql.{Date, Timestamp}
import java.util
import java.util.List
import java.util.stream.Collectors
-
import scala.collection.JavaConverters._
import scala.util.Random
@@ -133,7 +131,7 @@ class ColumnStatIndexTestBase extends
HoodieSparkClientTestBase {
params.operation.equals(DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)) &&
params.shouldValidateManually
validateColumnStatsIndex(
- params.testCase, params.metadataOpts,
params.expectedColStatsSourcePath, shouldValidateColumnStatsManually,
params.latestCompletedCommit)
+ params.testCase, params.metadataOpts,
params.expectedColStatsSourcePath, shouldValidateColumnStatsManually,
params.validationSortColumns)
}
}
@@ -231,29 +229,32 @@ class ColumnStatIndexTestBase extends
HoodieSparkClientTestBase {
columnsToIndex, writerSchemaOpt, maxBufferSize)
}
+ protected def validateColumnsToIndex(metaClient: HoodieTableMetaClient,
expectedColsToIndex: Seq[String]): Unit = {
+ val indexDefn =
metaClient.getIndexMetadata.get().getIndexDefinitions.get(PARTITION_NAME_COLUMN_STATS)
+ assertEquals(expectedColsToIndex, indexDefn.getSourceFields.asScala.toSeq)
+ }
+
+ protected def validateNonExistantColumnsToIndexDefn(metaClient:
HoodieTableMetaClient): Unit = {
+
assertTrue(!metaClient.getIndexMetadata.get().getIndexDefinitions.containsKey(PARTITION_NAME_COLUMN_STATS))
+ }
+
protected def validateColumnStatsIndex(testCase: ColumnStatsTestCase,
metadataOpts: Map[String, String],
expectedColStatsSourcePath: String,
validateColumnStatsManually: Boolean,
- latestCompletedCommit: String): Unit
= {
+ validationSortColumns: Seq[String]):
Unit = {
val metadataConfig = HoodieMetadataConfig.newBuilder()
.fromProperties(toProperties(metadataOpts))
.build()
val columnStatsIndex = new ColumnStatsIndexSupport(spark,
sourceTableSchema, metadataConfig, metaClient)
- val indexedColumns: Set[String] = HoodieTableMetadataUtil
+ val indexedColumnswithMeta: Set[String] = HoodieTableMetadataUtil
.getColumnsToIndex(metaClient.getTableConfig, metadataConfig,
convertScalaListToJavaList(sourceTableSchema.fieldNames)).asScala.toSet
+ val indexedColumns = indexedColumnswithMeta.filter(colName =>
!HoodieTableMetadataUtil.META_COL_SET_TO_INDEX.contains(colName))
val (expectedColStatsSchema, _) =
composeIndexSchema(sourceTableSchema.fieldNames, indexedColumns,
sourceTableSchema)
- val validationSortColumns = if (indexedColumns.contains("c5")) {
- Seq("c1_maxValue", "c1_minValue", "c2_maxValue", "c2_minValue",
"c3_maxValue",
- "c3_minValue", "c5_maxValue", "c5_minValue")
- } else {
- Seq("c1_maxValue", "c1_minValue", "c2_maxValue", "c2_minValue",
"c3_maxValue", "c3_minValue")
- }
-
- columnStatsIndex.loadTransposed(sourceTableSchema.fieldNames,
testCase.shouldReadInMemory) { transposedColStatsDF =>
+ columnStatsIndex.loadTransposed(indexedColumns.toSeq,
testCase.shouldReadInMemory) { transposedColStatsDF =>
// Match against expected column stats table
val expectedColStatsIndexTableDf =
spark.read
@@ -271,7 +272,7 @@ class ColumnStatIndexTestBase extends
HoodieSparkClientTestBase {
// TODO(HUDI-4557): support validation of column stats of avro log
files
// Collect Column Stats manually (reading individual Parquet files)
val manualColStatsTableDF =
- buildColumnStatsTableManually(basePath, sourceTableSchema.fieldNames,
sourceTableSchema.fieldNames, expectedColStatsSchema)
+ buildColumnStatsTableManually(basePath, indexedColumns.toSeq,
sourceTableSchema.fieldNames, expectedColStatsSchema)
assertEquals(asJson(sort(manualColStatsTableDF,
validationSortColumns)),
asJson(sort(transposedColStatsDF, validationSortColumns)))
@@ -380,5 +381,7 @@ object ColumnStatIndexTestBase {
latestCompletedCommit: String = null,
numPartitions: Integer = 4,
parquetMaxFileSize: Integer = 10 * 1024,
- smallFileLimit: Integer = 100 * 1024 * 1024)
+ smallFileLimit: Integer = 100 * 1024 * 1024,
+ validationSortColumns : Seq[String] =
Seq("c1_maxValue", "c1_minValue", "c2_maxValue",
+ "c2_minValue", "c3_maxValue",
"c3_minValue", "c5_maxValue", "c5_minValue"))
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
index 8a7c132136b..9ceadf5b420 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
@@ -24,7 +24,7 @@ import
org.apache.hudi.ColumnStatsIndexSupport.composeIndexSchema
import org.apache.hudi.DataSourceWriteOptions.{PARTITIONPATH_FIELD,
PRECOMBINE_FIELD, RECORDKEY_FIELD}
import org.apache.hudi.HoodieConversionUtils.toProperties
import org.apache.hudi.common.config.{HoodieCommonConfig,
HoodieMetadataConfig, HoodieStorageConfig}
-import org.apache.hudi.common.model.{HoodieBaseFile, HoodieFileGroup,
HoodieTableType}
+import org.apache.hudi.common.model.{HoodieBaseFile, HoodieFileGroup,
HoodieRecord, HoodieTableType}
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
import org.apache.hudi.common.testutils.HoodieTestUtils
import org.apache.hudi.common.util.ParquetUtils
@@ -62,6 +62,9 @@ import
scala.collection.convert.ImplicitConversions.`collection AsScalaIterable`
@Tag("functional")
class TestColumnStatsIndex extends ColumnStatIndexTestBase {
+ val DEFAULT_COLUMNS_TO_INDEX = Seq(HoodieRecord.COMMIT_TIME_METADATA_FIELD,
HoodieRecord.RECORD_KEY_METADATA_FIELD,
+ HoodieRecord.PARTITION_PATH_METADATA_FIELD,
"c1","c2","c3","c4","c5","c6","c7","c8")
+
@ParameterizedTest
@MethodSource(Array("testMetadataColumnStatsIndexParams"))
def testMetadataColumnStatsIndex(testCase: ColumnStatsTestCase): Unit = {
@@ -77,7 +80,8 @@ class TestColumnStatsIndex extends ColumnStatIndexTestBase {
DataSourceWriteOptions.TABLE_TYPE.key -> testCase.tableType.toString,
RECORDKEY_FIELD.key -> "c1",
PRECOMBINE_FIELD.key -> "c1",
- HoodieTableConfig.POPULATE_META_FIELDS.key -> "true"
+ HoodieTableConfig.POPULATE_META_FIELDS.key -> "true",
+ "hoodie.compact.inline.max.delta.commits" -> "10"
) ++ metadataOpts
doWriteAndValidateColumnStats(ColumnStatsTestParams(testCase,
metadataOpts, commonOpts,
@@ -94,7 +98,7 @@ class TestColumnStatsIndex extends ColumnStatIndexTestBase {
// NOTE: MOR and COW have different fixtures since MOR is bearing
delta-log files (holding
// deferred updates), diverging from COW
- val expectedColStatsSourcePath = if (testCase.tableType ==
HoodieTableType.COPY_ON_WRITE) {
+ var expectedColStatsSourcePath = if (testCase.tableType ==
HoodieTableType.COPY_ON_WRITE) {
"index/colstats/cow-updated2-column-stats-index-table.json"
} else {
"index/colstats/mor-updated2-column-stats-index-table.json"
@@ -105,6 +109,71 @@ class TestColumnStatsIndex extends ColumnStatIndexTestBase
{
expectedColStatsSourcePath = expectedColStatsSourcePath,
operation = DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
saveMode = SaveMode.Append))
+
+ validateColumnsToIndex(metaClient, DEFAULT_COLUMNS_TO_INDEX)
+
+ // update list of columns to explicit list of cols.
+ val metadataOpts1 = Map(
+ HoodieMetadataConfig.ENABLE.key -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key -> "true",
+ HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key() ->
"c1,c2,c3,c5,c6,c7,c8" // ignore c4
+ )
+
+ expectedColStatsSourcePath = if (testCase.tableType ==
HoodieTableType.COPY_ON_WRITE) {
+ "index/colstats/cow-updated3-column-stats-index-table.json"
+ } else {
+ "index/colstats/mor-updated3-column-stats-index-table.json"
+ }
+
+ doWriteAndValidateColumnStats(ColumnStatsTestParams(testCase,
metadataOpts1, commonOpts,
+ dataSourcePath = "index/colstats/update5-input-table-json",
+ expectedColStatsSourcePath = expectedColStatsSourcePath,
+ operation = DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
+ saveMode = SaveMode.Append))
+
+ validateColumnsToIndex(metaClient,
Seq(HoodieRecord.COMMIT_TIME_METADATA_FIELD,
HoodieRecord.RECORD_KEY_METADATA_FIELD,
+ HoodieRecord.PARTITION_PATH_METADATA_FIELD,
"c1","c2","c3","c5","c6","c7","c8"))
+
+ // lets explicitly override again. ignore c6
+ // update list of columns to explicit list of cols.
+ val metadataOpts2 = Map(
+ HoodieMetadataConfig.ENABLE.key -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key -> "true",
+ HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key() ->
"c1,c2,c3,c5,c7,c8" // ignore c4,c6
+ )
+
+ expectedColStatsSourcePath = if (testCase.tableType ==
HoodieTableType.COPY_ON_WRITE) {
+ "index/colstats/cow-updated4-column-stats-index-table.json"
+ } else {
+ "index/colstats/mor-updated4-column-stats-index-table.json"
+ }
+
+ doWriteAndValidateColumnStats(ColumnStatsTestParams(testCase,
metadataOpts2, commonOpts,
+ dataSourcePath = "index/colstats/update6-input-table-json",
+ expectedColStatsSourcePath = expectedColStatsSourcePath,
+ operation = DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
+ saveMode = SaveMode.Append))
+
+ validateColumnsToIndex(metaClient,
Seq(HoodieRecord.COMMIT_TIME_METADATA_FIELD,
HoodieRecord.RECORD_KEY_METADATA_FIELD,
+ HoodieRecord.PARTITION_PATH_METADATA_FIELD,
"c1","c2","c3","c5","c7","c8"))
+
+ // update list of columns to explicit list of cols.
+ val metadataOpts3 = Map(
+ HoodieMetadataConfig.ENABLE.key -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key -> "false",
+ HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key() ->
"c1,c2,c3,c5,c7" // ignore c4,c5,c8.
+ )
+ // disable col stats
+ doWriteAndValidateColumnStats(ColumnStatsTestParams(testCase,
metadataOpts3, commonOpts,
+ dataSourcePath = "index/colstats/update6-input-table-json",
+ expectedColStatsSourcePath = expectedColStatsSourcePath,
+ operation = DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
+ saveMode = SaveMode.Append,
+ shouldValidate = false,
+ shouldValidateManually = false))
+
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+ validateNonExistantColumnsToIndexDefn(metaClient)
}
/**
@@ -136,6 +205,9 @@ class TestColumnStatsIndex extends ColumnStatIndexTestBase {
expectedColStatsSourcePath = "index/colstats/cow-table-nested.json",
operation = DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
saveMode = SaveMode.Overwrite), true)
+
+ validateColumnsToIndex(metaClient,
Seq(HoodieRecord.COMMIT_TIME_METADATA_FIELD,
HoodieRecord.RECORD_KEY_METADATA_FIELD,
+ HoodieRecord.PARTITION_PATH_METADATA_FIELD,
"c1","c2","c3","c4","c5","c6","c7","c8"))
}
@ParameterizedTest
@@ -249,6 +321,8 @@ class TestColumnStatsIndex extends ColumnStatIndexTestBase {
numPartitions = 1,
parquetMaxFileSize = 100 * 1024 * 1024,
smallFileLimit = 0))
+
+ validateColumnsToIndex(metaClient, DEFAULT_COLUMNS_TO_INDEX)
}
@ParameterizedTest
@@ -325,6 +399,8 @@ class TestColumnStatsIndex extends ColumnStatIndexTestBase {
metaClient = HoodieTableMetaClient.reload(metaClient)
assertTrue(metaClient.getActiveTimeline.getRollbackTimeline.countInstants() > 0)
+
+ validateColumnsToIndex(metaClient, DEFAULT_COLUMNS_TO_INDEX)
}
def simulateFailureForLatestCommit(tableType: HoodieTableType, partitionCol:
String) : Unit = {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala
index 053446fb7c2..e560c8059fd 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala
@@ -91,7 +91,9 @@ class TestColumnStatsIndexWithSQL extends
ColumnStatIndexTestBase {
DataSourceReadOptions.ENABLE_DATA_SKIPPING.key -> "true",
DataSourceReadOptions.QUERY_TYPE.key ->
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL
) ++ metadataOpts
- setupTable(testCase, metadataOpts, commonOpts, shouldValidate = true,
useShortSchema = true)
+ setupTable(testCase, metadataOpts, commonOpts, shouldValidate = true,
useShortSchema = true,
+ validationSortColumns = Seq("c1_maxValue", "c1_minValue", "c2_maxValue",
+ "c2_minValue", "c3_maxValue", "c3_minValue"))
}
@ParameterizedTest
@@ -341,7 +343,9 @@ class TestColumnStatsIndexWithSQL extends
ColumnStatIndexTestBase {
}
private def setupTable(testCase: ColumnStatsTestCase, metadataOpts:
Map[String, String], commonOpts: Map[String, String],
- shouldValidate: Boolean, useShortSchema: Boolean =
false): Unit = {
+ shouldValidate: Boolean, useShortSchema: Boolean =
false,
+ validationSortColumns : Seq[String] =
Seq("c1_maxValue", "c1_minValue", "c2_maxValue",
+ "c2_minValue", "c3_maxValue", "c3_minValue",
"c5_maxValue", "c5_minValue")): Unit = {
val filePostfix = if (useShortSchema) {
"-short-schema"
} else {
@@ -352,14 +356,16 @@ class TestColumnStatsIndexWithSQL extends
ColumnStatIndexTestBase {
expectedColStatsSourcePath =
s"index/colstats/column-stats-index-table${filePostfix}.json",
operation = DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
shouldValidateManually = !useShortSchema,
- saveMode = SaveMode.Overwrite))
+ saveMode = SaveMode.Overwrite,
+ validationSortColumns = validationSortColumns))
doWriteAndValidateColumnStats(ColumnStatsTestParams(testCase,
metadataOpts, commonOpts,
dataSourcePath = "index/colstats/another-input-table-json",
expectedColStatsSourcePath =
s"index/colstats/updated-column-stats-index-table${filePostfix}.json",
operation = DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
shouldValidateManually = !useShortSchema,
- saveMode = SaveMode.Append))
+ saveMode = SaveMode.Append,
+ validationSortColumns = validationSortColumns))
// NOTE: MOR and COW have different fixtures since MOR is bearing
delta-log files (holding
// deferred updates), diverging from COW
@@ -375,7 +381,8 @@ class TestColumnStatsIndexWithSQL extends
ColumnStatIndexTestBase {
operation = DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
saveMode = SaveMode.Append,
shouldValidate,
- shouldValidateManually = !useShortSchema))
+ shouldValidateManually = !useShortSchema,
+ validationSortColumns = validationSortColumns))
}
def verifyFileIndexAndSQLQueries(opts: Map[String, String],
isTableDataSameAsAfterSecondInstant: Boolean = false, verifyFileCount: Boolean
= true): Unit = {
