cshuo commented on code in PR #18372:
URL: https://github.com/apache/hudi/pull/18372#discussion_r3542089238
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1585,102 +1106,15 @@ public BatchMetadataConversionFunction(String
instantTime, HoodieCommitMetadata
}
@Override
- public Map<String, HoodieData<HoodieRecord>> convertMetadata() {
- Map<String, HoodieData<HoodieRecord>> partitionToRecordMap =
- HoodieMetadataWriteUtils.convertMetadataToRecords(
- engineContext, dataWriteConfig, commitMetadata, instantTime,
dataMetaClient, getTableMetadata(),
- dataWriteConfig.getMetadataConfig(),
- partitionsToUpdate, dataWriteConfig.getBloomFilterType(),
- dataWriteConfig.getBloomIndexParallelism(),
dataWriteConfig.getWritesFileIdEncoding(), getEngineType(),
- Option.of(dataWriteConfig.getRecordMerger().getRecordType()));
-
- // Updates for record index are created by parsing the WriteStatus which
is a hudi-client object. Hence, we cannot yet move this code
- // to the HoodieTableMetadataUtil class in hudi-common.
- if (partitionsToUpdate.contains(RECORD_INDEX.getPartitionPath())) {
- HoodieData<HoodieRecord> additionalUpdates =
getRecordIndexAdditionalUpserts(partitionToRecordMap.get(RECORD_INDEX.getPartitionPath()),
commitMetadata);
- partitionToRecordMap.put(RECORD_INDEX.getPartitionPath(),
partitionToRecordMap.get(RECORD_INDEX.getPartitionPath()).union(additionalUpdates));
- }
- if (partitionsToUpdate.stream().anyMatch(partition ->
partition.startsWith(EXPRESSION_INDEX.getPartitionPath()))) {
- updateExpressionIndexIfPresent(commitMetadata, instantTime,
partitionToRecordMap);
- }
- if (partitionsToUpdate.stream().anyMatch(partition ->
partition.startsWith(SECONDARY_INDEX.getPartitionPath()))) {
- updateSecondaryIndexIfPresent(commitMetadata, partitionToRecordMap,
instantTime);
- }
- return partitionToRecordMap;
- }
- }
-
- /**
- * Update expression index from {@link HoodieCommitMetadata}.
- */
- private void updateExpressionIndexIfPresent(HoodieCommitMetadata
commitMetadata, String instantTime,
- Map<String,
HoodieData<HoodieRecord>> partitionToRecordMap) {
- if
(!MetadataPartitionType.EXPRESSION_INDEX.isMetadataPartitionAvailable(dataMetaClient))
{
- return;
- }
- dataMetaClient.getTableConfig().getMetadataPartitions()
- .stream()
- .filter(partition ->
partition.startsWith(HoodieTableMetadataUtil.PARTITION_NAME_EXPRESSION_INDEX_PREFIX))
- .forEach(partition -> {
- HoodieData<HoodieRecord> expressionIndexRecords;
- try {
- expressionIndexRecords = getExpressionIndexUpdates(commitMetadata,
partition, instantTime);
- } catch (Exception e) {
- throw new HoodieMetadataException(String.format("Failed to get
expression index updates for partition %s", partition), e);
- }
- partitionToRecordMap.put(partition, expressionIndexRecords);
- });
- }
-
- /**
- * Loads the file slices touched by the commit due to given instant time and
returns the records for the expression index.
- *
- * @param commitMetadata {@code HoodieCommitMetadata}
- * @param indexPartition partition name of the expression index
- * @param instantTime timestamp at of the current update commit
- */
- protected HoodieData<HoodieRecord>
getExpressionIndexUpdates(HoodieCommitMetadata commitMetadata, String
indexPartition, String instantTime) throws Exception {
- throw new UnsupportedOperationException("Expression Index only supported
with SPARK engine.");
- }
-
- private void updateSecondaryIndexIfPresent(HoodieCommitMetadata
commitMetadata, Map<String, HoodieData<HoodieRecord>> partitionToRecordMap,
- String instantTime) {
- boolean secondaryIndexMetadataPartitionAvailable =
SECONDARY_INDEX.isMetadataPartitionAvailable(dataMetaClient);
- if (!secondaryIndexMetadataPartitionAvailable) {
- return;
- }
- // If write operation type based on commit metadata is COMPACT or CLUSTER
then no need to update,
- // because these operations do not change the secondary key - record key
mapping.
- WriteOperationType operationType = commitMetadata.getOperationType();
- if (operationType.isInsertOverwriteOrDeletePartition()) {
- throw new HoodieIndexException(String.format("Can not perform operation
%s on secondary index", operationType));
- } else if (operationType == WriteOperationType.COMPACT || operationType ==
WriteOperationType.CLUSTER) {
- return;
+ public List<IndexPartitionAndRecords> convertMetadata() {
+ return partitionsToUpdate.stream().flatMap(indexPartition -> {
+ MetadataPartitionType partitionType =
MetadataPartitionType.fromPartitionPath(indexPartition);
Review Comment:
Fixed.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/model/FileSliceAndPartition.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.metadata.model;
+
+import org.apache.hudi.common.model.FileSlice;
+
+import lombok.AllArgsConstructor;
+import lombok.ToString;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+
+/**
+ * Holder for a {@link FileSlice} and its partition path.
+ */
+@AllArgsConstructor
+@Getter
+@Setter
+@ToString
Review Comment:
Fixed.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestGlobalRecordLevelIndexTableVersionSix.scala:
##########
@@ -18,15 +18,49 @@
package org.apache.hudi.functional
-import org.apache.hudi.common.table.HoodieTableConfig
+import org.apache.hudi.DataSourceWriteOptions
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.model.HoodieTableType
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator
import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.metadata.MetadataPartitionType
+import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Tag
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.EnumSource
+
+import java.util.Collections
@Tag("functional-b")
class TestGlobalRecordLevelIndexTableVersionSix extends
TestGlobalRecordLevelIndex {
override def commonOpts: Map[String, String] = super.commonOpts ++ Map(
HoodieTableConfig.VERSION.key() -> "6",
HoodieWriteConfig.WRITE_TABLE_VERSION.key() -> "6"
)
+
+ @ParameterizedTest
+ @EnumSource(classOf[HoodieTableType])
+ override def testRLIUpsertAndDropIndex(tableType: HoodieTableType): Unit = {
+ val hudiOpts = commonOpts ++ Map(DataSourceWriteOptions.TABLE_TYPE.key ->
tableType.name(),
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key -> "true")
+ doWriteAndValidateDataAndRecordIndex(hudiOpts,
+ operation = DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
+ saveMode = SaveMode.Overwrite)
+
+ val writeConfig = getWriteConfig(hudiOpts)
+ writeConfig.setSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA)
+
getHoodieWriteClient(writeConfig).dropIndex(Collections.singletonList(MetadataPartitionType.RECORD_INDEX.getPartitionPath))
+ assertEquals(0, getFileGroupCountForRecordIndex(writeConfig))
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+ assertEquals(0,
metaClient.getTableConfig.getMetadataPartitionsInflight.size())
+ // only files, col stats, partition stats partition should be present.
+ assertEquals(2, metaClient.getTableConfig.getMetadataPartitions.size())
+
Review Comment:
Fixed.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/expression/TestExpressionIndexer.java:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.
+ */
Review Comment:
Fixed.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/record/TestPartitionedRecordIndexer.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.
+ */
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
