vinishjail97 commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3974592001
##########
hudi-common/src/main/java/org/apache/hudi/common/util/FileFormatUtils.java:
##########
@@ -252,6 +266,22 @@ public abstract Map<String, String>
readFooter(HoodieStorage storage, boolean re
*/
public abstract Set<Pair<String, Long>> filterRowKeys(HoodieStorage storage,
StoragePath filePath, Set<String> filter);
+ /**
+ * Read the rowKey list matching the given filter, from the given data file.
+ * If the filter is empty, then this will return all the row keys and
corresponding positions.
+ * Formats that support data files written outside Hudi override this method
to key every row of such a file,
+ * which carries no record key, by the file path relative to the table base
path and the row position.
+ *
+ * @param storage {@link HoodieStorage} instance.
+ * @param filePath the data file path.
+ * @param basePath the table base path.
+ * @param filter record keys filter.
+ * @return set of pairs of row key and position matching candidateRecordKeys.
+ */
+ public Set<Pair<String, Long>> filterRowKeys(HoodieStorage storage,
StoragePath filePath, StoragePath basePath, Set<String> filter) {
+ return filterRowKeys(storage, filePath, filter);
Review Comment:
Applied in 9085866, and the new `getRowKeyIterator` default throws the same
`UnsupportedOperationException`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java:
##########
@@ -258,11 +260,29 @@ protected static <T> HoodieData<HoodieRecord>
readRecordKeysFromFileSliceSnapsho
engineContext.setJobStatus(activeModule, "Record Index: reading record
keys from " + fileSlices.size() + " file slices");
final int parallelism = Math.min(fileSlices.size(),
recordIndexMaxParallelism);
+ // a table without record keys, e.g. one that registers files written
outside Hudi, keys every row by the path of
+ // its base file relative to the table and the row position, the same way
the record index update path does
+ final boolean generateRecordKeys =
!metaClient.getTableConfig().hasRecordKey();
+ final int fileIdEncoding = dataWriteConfig.getWritesFileIdEncoding();
ReaderContextFactory<T> readerContextFactory =
engineContext.getReaderContextFactory(metaClient);
return engineContext.parallelize(fileSlices,
parallelism).flatMap(partitionAndFileSlice -> {
final String partition = partitionAndFileSlice.getPartitionPath();
final FileSlice fileSlice = partitionAndFileSlice.getFileSlice();
final String fileId = fileSlice.getFileId();
+ long baseFileInstantTimeMillis =
HoodieMetadataPayload.parseRecordIndexInstantTime(fileSlice.getBaseInstantTime());
+ if (generateRecordKeys) {
Review Comment:
Added the unit tests in 9085866: `TestRecordIndexer` covers the initialize
guard through `readRecordKeysFromFileSliceSnapshot`, and
`TestSecondaryIndexRecordGenerationUtils` covers the secondary index twin
through `getRecordKeyToSecondaryKey`, both with a log file on a `hasRecordKey()
== false` slice. I left the up-front enablement gate out: a table converted
from another format never has log files, so the guard is a consistency check
rather than a user-facing configuration error, and rejecting the index configs
would also block the metadata table for tables that already carry them.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/secondary/TestSecondaryIndexer.java:
##########
@@ -171,6 +176,51 @@ void testBuildUpdateForCompactReturnsEmpty() {
commitMetadata)).isEmpty());
}
+ @Test
+ void testBuildUpdateForReplaceCommitFromExternalWriterWithoutWriteStats() {
+ // files written outside Hudi are registered through replace commits
without a known operation type. A fresh
+ // HoodieCommitMetadata carries UNKNOWN; a writer may also leave the type
null. A commit that only drops files has
+ // no write stats but still removes the records of the replaced file
groups from the index.
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(getDefaultStorageConf());
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ HoodieIndexMetadata indexMetadata = mock(HoodieIndexMetadata.class);
+ HoodieIndexDefinition indexDefinition = mock(HoodieIndexDefinition.class);
+
+ when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig);
+ when(metaClient.getIndexMetadata()).thenReturn(Option.of(indexMetadata));
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+
when(metaClient.getIndexForMetadataPartition("secondary_index_idx")).thenReturn(Option.of(indexDefinition));
+
when(tableConfig.getMetadataPartitions()).thenReturn(Collections.singleton("secondary_index_idx"));
+
when(indexMetadata.getIndexDefinitions()).thenReturn(Collections.singletonMap("secondary_index_idx",
indexDefinition));
+ when(indexDefinition.getIndexName()).thenReturn("secondary_index_idx");
+
+ HoodieReplaceCommitMetadata commitMetadata = new
HoodieReplaceCommitMetadata();
+ commitMetadata.setOperationType(null);
+ commitMetadata.addReplaceFileId("p1", "file_1.parquet");
+
+ HoodieData<HoodieRecord> deletes =
engineContext.parallelize(Collections.singletonList(
+
HoodieMetadataPayload.createSecondaryIndexRecord("p1/file_1.parquet_0",
"alice", "secondary_index_idx", true)), 1);
+ try (MockedStatic<SecondaryIndexRecordGenerationUtils>
mockedGenerationUtils = mockStatic(SecondaryIndexRecordGenerationUtils.class)) {
+ mockedGenerationUtils.when(() ->
SecondaryIndexRecordGenerationUtils.convertWriteStatsToSecondaryIndexRecords(
+ eq(Collections.emptyList()), eq("016"), eq(indexDefinition),
eq(metadataConfig), eq(metaClient), eq(engineContext), eq(writeConfig),
eq(commitMetadata)))
+ .thenReturn(deletes);
+
+ SecondaryIndexer indexer = new SecondaryIndexer(engineContext,
writeConfig, metaClient);
+ List<IndexPartitionAndRecords> result =
indexer.buildUpdate(IndexUpdateContext.of(
+ "016",
+ mock(HoodieBackedTableMetadata.class),
+ Lazy.lazily(() -> mock(HoodieTableFileSystemView.class)),
+ commitMetadata));
+
+ assertEquals(1, result.size());
+ assertEquals("secondary_index_idx", result.get(0).indexPartitionName());
+ assertEquals(1, result.get(0).indexRecords().collectAsList().size());
+ }
Review Comment:
Done in 9085866. The test is parameterized over the replace map: with a
replaced file id the stubbed deletes come back, with an empty map the result is
empty.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/partitionstats/PartitionStatsIndexer.java:
##########
@@ -114,7 +114,7 @@ public List<IndexPartitionAndRecords>
buildUpdate(IndexUpdateContext context) {
checkState(MetadataPartitionType.COLUMN_STATS.isMetadataPartitionAvailable(dataTableMetaClient),
"Column stats partition must be enabled to generate partition stats.
Please enable: " +
HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key());
// Generate Hoodie Pair data of partition name and list of column range
metadata for all the files in that partition
- boolean isDeletePartition =
context.commitMetadata().getOperationType().equals(WriteOperationType.DELETE_PARTITION);
+ boolean isDeletePartition =
WriteOperationType.DELETE_PARTITION.equals(context.commitMetadata().getOperationType());
Review Comment:
Done in 9085866, parameterized with `@NullSource` and `@EnumSource(names =
"UPSERT")`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -258,7 +318,8 @@ public static <T> HoodieData<HoodieRecord>
readSecondaryKeysFromFileSlices(Hoodi
return engineContext.parallelize(fileSlices,
parallelism).flatMap(partitionAndBaseFile -> {
final String partition = partitionAndBaseFile.getPartitionPath();
Review Comment:
Removed both in 9085866.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/BaseFileRecordParsingUtils.java:
##########
@@ -65,16 +69,35 @@ public static Iterator<HoodieRecord>
generateRLIMetadataHoodieRecordsForBaseFile
String instantTime,
HoodieStorage storage,
boolean isPartitionedRLI) {
+ return generateRLIMetadataHoodieRecordsForBaseFile(basePath, writeStat,
writesFileIdEncoding, instantTime, storage, isPartitionedRLI, false);
Review Comment:
Removed in 9085866; the three test call sites pass `false`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -205,6 +227,44 @@ public static <T> HoodieData<HoodieRecord>
convertWriteStatsToSecondaryIndexReco
return HoodieTableMetadataUtil.reduceByKeys(secondaryIndexRecords,
parallelism, false);
}
+ /**
+ * Generates delete records for every record of the file groups that the
given replace commit replaces without
+ * writing to them again, e.g. files written outside Hudi that are
superseded by newer files. Records that the same
Review Comment:
Reworded in 9085866. The javadoc now names the keyed-table case for the
`reduceByKeys` precedence and states that a keyless table never writes a file
group it replaces.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.functional;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieDeltaWriteStat;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ExternalFilePathUtil;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.data.HoodieJavaRDD;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.testutils.HoodieClientTestBase;
+
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
+import static org.apache.hudi.index.HoodieIndex.IndexType.INMEMORY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Asserts that the record index and the secondary index are maintained for
parquet files written outside Hudi
+ * and registered in the table through replace commits, the way tables
converted from other formats are.
+ * Such a table has no record key, so every row is keyed by the file path
relative to the table and the row position.
+ * Some writers, e.g. Paimon, place their files in a directory below the
partition; that prefix is part of the file id.
+ */
+public class TestExternalFileRecordAndSecondaryIndex extends
HoodieClientTestBase {
+
+ private static final String ID_FIELD = "id";
+ private static final String NAME_FIELD = "name";
+ private static final String PARTITION = "americas/brazil";
+ private static final String PREFIX = "bucket-0";
+ private static final HoodieSchema SCHEMA =
HoodieSchema.createRecord("external", null, null, false, Arrays.asList(
+ HoodieSchemaField.of(ID_FIELD,
HoodieSchema.create(HoodieSchemaType.INT)),
+ HoodieSchemaField.of(NAME_FIELD,
HoodieSchema.create(HoodieSchemaType.STRING))));
+
+ /** Partitioned and unpartitioned tables, with the files placed directly in
the partition or below a prefix. */
+ static Stream<Arguments> partitionsAndPrefixes() {
+ return Stream.of(
+ Arguments.of(PARTITION, Option.empty()),
+ Arguments.of("", Option.empty()),
+ Arguments.of(PARTITION, Option.of(PREFIX)),
+ Arguments.of("", Option.of(PREFIX)));
+ }
+
+ /** The shapes above with a global record index, plus a partitioned record
index, which the secondary index does not build on. */
+ static Stream<Arguments> partitionsPrefixesAndRecordIndexKinds() {
+ return Stream.concat(
+ partitionsAndPrefixes().map(arguments ->
Arguments.of(arguments.get()[0], arguments.get()[1], false)),
+ Stream.of(Arguments.of(PARTITION, Option.empty(), true)));
+ }
+
+ @ParameterizedTest
+ @MethodSource("partitionsPrefixesAndRecordIndexKinds")
+ public void testRecordAndSecondaryIndexForExternalFiles(String
partitionPath, Option<String> prefix, boolean partitionedRecordIndex) throws
Exception {
+ initExternalTable();
+ HoodieWriteConfig writeConfig = writeConfig(true, partitionedRecordIndex);
+ writeClient = getHoodieWriteClient(writeConfig);
+ writeClient.setOperationType(WriteOperationType.UNKNOWN);
+ Option<String> recordIndexPartition = partitionedRecordIndex ?
Option.of(partitionPath) : Option.empty();
+
+ // first commit registers one file with three rows
+ ExternalFile file1 = new ExternalFile(partitionPath, prefix,
"file_1.parquet");
+ commitReplace(Collections.singletonList(Pair.of(file1, rows(1, "alice", 2,
"bob", 3, "alice"))), Collections.emptyMap());
+
+ HoodieBackedTableMetadata tableMetadata = tableMetadata(writeConfig);
+ assertRecordIndex(tableMetadata, recordIndexPartition, file1, 3);
+ if (!partitionedRecordIndex) {
+ assertEquals(mapOf("alice", setOf(file1.key(0), file1.key(2)), "bob",
setOf(file1.key(1))),
+ readSecondaryIndex(tableMetadata, secondaryIndexPartition(),
Arrays.asList("alice", "bob", "carol")));
+ }
+
+ // second commit replaces the file with one that keeps bob and adds carol
+ ExternalFile file2 = new ExternalFile(partitionPath, prefix,
"file_2.parquet");
+ commitReplace(Collections.singletonList(Pair.of(file2, rows(2, "bob", 4,
"carol"))),
+ Collections.singletonMap(partitionPath,
Collections.singletonList(file1.fileId())));
+
+ tableMetadata = tableMetadata(writeConfig);
+ assertTrue(readRecordIndex(tableMetadata, recordIndexPartition,
file1.keys(3)).isEmpty());
+ assertRecordIndex(tableMetadata, recordIndexPartition, file2, 2);
+ if (!partitionedRecordIndex) {
+ assertEquals(mapOf("bob", setOf(file2.key(0)), "carol",
setOf(file2.key(1))),
+ readSecondaryIndex(tableMetadata, secondaryIndexPartition(),
Arrays.asList("alice", "bob", "carol")));
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("partitionsAndPrefixes")
+ public void testIndexesAreBuiltFromRegisteredFilesWhenEnabledLater(String
partitionPath, Option<String> prefix) throws Exception {
+ initExternalTable();
+ // the first file is registered while both indexes are off
+ writeClient = getHoodieWriteClient(writeConfig(false, false));
+ writeClient.setOperationType(WriteOperationType.UNKNOWN);
+ ExternalFile file1 = new ExternalFile(partitionPath, prefix,
"file_1.parquet");
+ commitReplace(Collections.singletonList(Pair.of(file1, rows(1, "alice", 2,
"bob", 3, "alice"))), Collections.emptyMap());
+
+ // the second file is registered with both indexes on, which first builds
them from the registered file
+ HoodieWriteConfig indexedWriteConfig = writeConfig(true, false);
+ writeClient = getHoodieWriteClient(indexedWriteConfig);
+ writeClient.setOperationType(WriteOperationType.UNKNOWN);
+ ExternalFile file2 = new ExternalFile(partitionPath, prefix,
"file_2.parquet");
+ commitReplace(Collections.singletonList(Pair.of(file2, rows(4, "carol"))),
Collections.emptyMap());
+
+ HoodieBackedTableMetadata tableMetadata =
tableMetadata(indexedWriteConfig);
+ assertRecordIndex(tableMetadata, Option.empty(), file1, 3);
+ assertRecordIndex(tableMetadata, Option.empty(), file2, 1);
+ String secondaryIndexPartition = secondaryIndexPartition();
+ assertEquals(mapOf("alice", setOf(file1.key(0), file1.key(2)), "bob",
setOf(file1.key(1)), "carol", setOf(file2.key(0))),
+ readSecondaryIndex(tableMetadata, secondaryIndexPartition,
Arrays.asList("alice", "bob", "carol")));
+
+ // the third commit only drops the first file
+ commitReplace(Collections.emptyList(),
Collections.singletonMap(partitionPath,
Collections.singletonList(file1.fileId())));
+
+ tableMetadata = tableMetadata(indexedWriteConfig);
+ assertTrue(readRecordIndex(tableMetadata, Option.empty(),
file1.keys(3)).isEmpty());
+ assertRecordIndex(tableMetadata, Option.empty(), file2, 1);
+ assertEquals(mapOf("carol", setOf(file2.key(0))),
+ readSecondaryIndex(tableMetadata, secondaryIndexPartition,
Arrays.asList("alice", "bob", "carol")));
+ }
+
+ /** Tables registered from other formats carry neither meta fields nor
record key fields. */
+ private void initExternalTable() throws IOException {
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty(HoodieTableConfig.POPULATE_META_FIELDS.key(),
"false");
+ tableProperties.setProperty(HoodieTableConfig.RECORDKEY_FIELDS.key(), "");
+ initMetaClient(tableProperties);
+ }
+
+ private HoodieWriteConfig writeConfig(boolean withIndexes, boolean
partitionedRecordIndex) {
+ HoodieMetadataConfig.Builder metadataConfig =
HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withMetadataIndexColumnStats(false);
+ if (withIndexes && partitionedRecordIndex) {
+ // the secondary index builds on the global record index only
+ metadataConfig.withEnableRecordLevelIndex(true);
+ } else if (withIndexes) {
+ metadataConfig.withEnableGlobalRecordLevelIndex(true)
+ .withSecondaryIndexEnabled(true)
+ .withSecondaryIndexForColumn(NAME_FIELD);
+ }
+ return HoodieWriteConfig.newBuilder()
+ .withPath(metaClient.getBasePath())
+ .withSchema(SCHEMA.toString())
+ .withPopulateMetaFields(false)
+ // file ids of external files are file names, not UUIDs, so the record
index stores them as raw strings
+ .withWritesFileIdEncoding(1)
+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(INMEMORY).build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMetadataConfig(metadataConfig.build())
+ .build();
+ }
+
+ /**
+ * Registers the given files in one replace commit, the way an external
writer does: the files are written to
+ * their location below the table, and the commit records them with the
external file marker.
+ */
+ private void commitReplace(List<Pair<ExternalFile, List<Pair<Integer,
String>>>> newFiles,
+ Map<String, List<String>>
partitionToReplacedFileIds) throws IOException {
+ String instantTime =
writeClient.startCommit(HoodieTimeline.REPLACE_COMMIT_ACTION, metaClient);
+ List<WriteStatus> writeStatuses = new ArrayList<>();
+ for (Pair<ExternalFile, List<Pair<Integer, String>>> fileAndRows :
newFiles) {
+ ExternalFile file = fileAndRows.getLeft();
+ long fileSize = writeParquetFile(new
Path(metaClient.getBasePath().toString(), file.relativePath()),
fileAndRows.getRight());
+ WriteStatus writeStatus = new WriteStatus();
+ writeStatus.setFileId(file.fileId());
+ writeStatus.setPartitionPath(file.partitionPath);
+ HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat();
+ writeStat.setFileId(file.fileId());
+ writeStat.setPath(file.markedPath(instantTime));
+ writeStat.setPartitionPath(file.partitionPath);
+ writeStat.setNumWrites(fileAndRows.getRight().size());
+ writeStat.setNumInserts(fileAndRows.getRight().size());
+ writeStat.setTotalWriteBytes(fileSize);
+ writeStat.setFileSizeInBytes(fileSize);
+ writeStatus.setStat(writeStat);
+ writeStatuses.add(writeStatus);
+ }
+ metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime), Option.empty());
+ writeClient.commit(instantTime, jsc.parallelize(writeStatuses, 1),
Option.empty(),
+ HoodieTimeline.REPLACE_COMMIT_ACTION, partitionToReplacedFileIds);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ }
+
+ private long writeParquetFile(Path path, List<Pair<Integer, String>> rows)
throws IOException {
+ Configuration conf =
metaClient.getStorageConf().unwrapAs(Configuration.class);
+ try (ParquetWriter<GenericRecord> writer =
AvroParquetWriter.<GenericRecord>builder(path)
+ .withSchema(SCHEMA.toAvroSchema()).withConf(conf).build()) {
+ for (Pair<Integer, String> row : rows) {
+ GenericRecord record = new GenericData.Record(SCHEMA.toAvroSchema());
+ record.put(ID_FIELD, row.getLeft());
+ record.put(NAME_FIELD, row.getRight());
+ writer.write(record);
+ }
+ }
+ return path.getFileSystem(conf).getFileStatus(path).getLen();
+ }
+
+ private HoodieBackedTableMetadata tableMetadata(HoodieWriteConfig
writeConfig) {
+ return new HoodieBackedTableMetadata(context, metaClient.getStorage(),
writeConfig.getMetadataConfig(), writeConfig.getBasePath(), true);
+ }
+
+ private String secondaryIndexPartition() {
+ return
metaClient.getIndexMetadata().get().getIndexDefinitions().keySet().stream()
+ .filter(indexName ->
indexName.startsWith(HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX))
+ .findFirst().get();
+ }
+
+ /** Asserts that every row of the file is in the record index and points at
the file. */
+ private void assertRecordIndex(HoodieBackedTableMetadata tableMetadata,
Option<String> recordIndexPartition, ExternalFile file, int rowCount) {
+ Map<String, HoodieRecordGlobalLocation> locations =
readRecordIndex(tableMetadata, recordIndexPartition, file.keys(rowCount));
+ assertEquals(rowCount, locations.size());
+ locations.values().forEach(location -> {
+ assertEquals(file.partitionPath, location.getPartitionPath());
+ assertEquals(file.fileId(), location.getFileId());
+ });
+ }
+
+ private Map<String, HoodieRecordGlobalLocation>
readRecordIndex(HoodieBackedTableMetadata tableMetadata, Option<String>
recordIndexPartition,
+ List<String>
recordKeys) {
+ return
tableMetadata.readRecordIndexLocationsWithKeys(HoodieJavaRDD.of(jsc.parallelize(recordKeys,
1)), recordIndexPartition).collectAsList().stream()
Review Comment:
Fixed in 9085866. `RecordLevelIndexSupport.filterCandidateFiles` now takes
the pruned file slices and matches `fileSlice.getFileId` against the lookup
results, and `SecondaryIndexSupport` and both RLI supports go through it;
`getPrunedStoragePaths` is gone. The functional test reads the table through
Spark with `where(name = ...)` and asserts the rows and a `numFiles` of one out
of two registered files. That read runs on the unpartitioned shapes, because
the test harness declares a `partition_path` partition field that the schema of
the external files does not carry.
--
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]
