vinishjail97 commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3974590936
##########
hudi-common/src/test/java/org/apache/hudi/common/util/TestExternalFilePathUtil.java:
##########
@@ -230,4 +230,19 @@ public void testRoundTrip_WithPrefix() {
assertEquals(prefix + "/" + originalFile, parsed[0]);
assertEquals(COMMIT_TIME, parsed[1]);
}
+
+ @Test
+ public void testGetFilePathInPartition() {
Review Comment:
Restored both marker-form assertions alongside the nested case in 9085866,
thanks for catching the drop.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java:
##########
@@ -478,11 +478,48 @@ private HoodieData<HoodieRecord>
getRecordIndexAdditionalUpserts(
} else if (operationType == WriteOperationType.DELETE_PARTITION) {
// all records from the target partition(s) to be deleted from RLI
return getRecordIndexReplacedRecords((HoodieReplaceCommitMetadata)
commitMetadata, fsView);
Review Comment:
Filed the follow-up as https://github.com/apache/hudi/issues/19886, listing
the three items: the `HoodieAvroParquetReader.getRecordKeyIterator` fallback,
the `baseFile.getFileName()` path construction in `HoodieTableMetadataUtil` at
both call sites, and the shared `DELETE_PARTITION` path. Thanks for the extra
data points.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/BaseFileRecordParsingUtils.java:
##########
@@ -67,7 +69,9 @@ public static Iterator<HoodieRecord>
generateRLIMetadataHoodieRecordsForBaseFile
boolean isPartitionedRLI) {
String partition = writeStat.getPartitionPath();
String latestFileName = FSUtils.getFileNameFromPath(writeStat.getPath());
- String fileId = FSUtils.getFileId(latestFileName);
+ // a file written outside Hudi keeps its own name, which may contain
underscores, so the file id is parsed from the marker
+ String fileId =
FileNameParser.parseBaseFile(latestFileName).map(FileNameParser.BaseFileName::getFileId)
Review Comment:
Fixed in 9085866. `BaseRecordIndexer.getFileIdEncoding` returns
`RECORD_INDEX_FIELD_FILEID_ENCODING_RAW_STRING` whenever `hasRecordKey()` is
false, on both the update path and the initialize path, and the functional test
now runs on the default config with `withWritesFileIdEncoding` removed. The
unit test also drops its encoding stub, so it would fail if the derivation went
away.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -205,6 +221,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:
Kept the signature as is. The parameters are the same set every other
generation helper in this class takes, and a context object for one private
method would add a type without a second caller. Happy to revisit if the class
grows another caller.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/secondary/SecondaryIndexer.java:
##########
@@ -115,7 +116,7 @@ public List<IndexPartitionAndRecords>
buildUpdate(IndexUpdateContext context) {
// 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 =
context.commitMetadata().getOperationType();
- if (operationType.isInsertOverwriteOrDeletePartition()) {
+ if (operationType != null &&
operationType.isInsertOverwriteOrDeletePartition()) {
throw new HoodieIndexException(String.format("Can not perform operation
%s on secondary index", operationType));
} else if (operationType == WriteOperationType.COMPACT || operationType ==
WriteOperationType.CLUSTER) {
Review Comment:
Fixed in 9085866. Both indexers now call
`BaseIndexer.checkClusteringKeepsRecordKeys`, which fails a CLUSTER commit on a
table without record keys before any early return. Unit tests cover both
indexers, and the functional test replays a clustering commit through the batch
metadata path, which is the path `completeClustering` uses. Running the
clustering executor itself on such a table fails earlier today while it
instantiates `SimpleKeyGenerator` without a record key field, so the functional
case cannot reach the guard that way.
##########
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");
Review Comment:
Took the `checkState` option in 9085866, and did not add the re-registration
commit, because the scenario cannot produce a valid table:
`AbstractTableFileSystemView.getLatestBaseFile` returns empty for a replaced
file group even when the same commit writes it again, so the file would
disappear from the snapshot while its keys survive in the indexes.
`BaseRecordIndexer.checkReplacedFileGroupsAreNotWritten` now rejects a keyless
replace commit whose written file ids overlap its replaced file ids, with a
unit test, and the javadoc on both helpers says why. The RLI anti-join and the
SI `writtenFileIds` filter stay for keyed tables that take the unknown-op
branch.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/record/TestPartitionedRecordIndexer.java:
##########
@@ -198,7 +198,7 @@ void
testBuildUpdateWithNonEmptyCommitMetadataProducesPartitionEntry() {
mockedMetadataUtil.when(() ->
HoodieTableMetadataUtil.reduceByKeys(any(), anyInt(), anyBoolean()))
.thenAnswer(invocation -> invocation.getArgument(0));
mockedBaseFileParsingUtils.when(() -> BaseFileRecordParsingUtils
- .generateRLIMetadataHoodieRecordsForBaseFile(any(), any(),
any(), any(), any(), anyBoolean()))
+ .generateRLIMetadataHoodieRecordsForBaseFile(any(), any(),
any(), any(), any(), anyBoolean(), anyBoolean()))
Review Comment:
Fixed in 9085866. The test stubs `hasRecordKey()` to true and pins
`eq(false)` in both the stub and the verify, like `TestRecordIndexer`.
##########
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) {
+ checkState(fileSlice.getBaseFile().isPresent() &&
!fileSlice.hasLogFiles(),
+ "File group " + fileId + " in partition " + partition + " needs a
base file and no log files to key its rows by "
+ + "position, because the table has no record key");
+ StoragePath dataFilePath =
fileSlice.getBaseFile().get().getStoragePath();
+ HoodieStorage storage = metaClient.getStorage();
+ Set<String> recordKeys = HoodieIOFactory.getIOFactory(storage)
+
.getFileFormatUtils(metaClient.getTableConfig().getBaseFileFormat())
+ .readRowKeys(storage, dataFilePath, metaClient.getBasePath());
Review Comment:
Fixed in 9085866. `FileFormatUtils.getRowKeyIterator(storage, filePath,
basePath)` streams the positional keys, `ParquetUtils` implements it and builds
its set-based `filterRowKeys` on top of it, and the initialize path maps the
iterator through `CloseableMappingIterator` instead of collecting a `Set`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -107,7 +115,13 @@ public static <T> HoodieData<HoodieRecord>
convertWriteStatsToSecondaryIndexReco
HoodieSchema tableSchema;
try {
- tableSchema = tryResolveSchemaForTable(dataMetaClient).get();
+ // a table without any completed commit, e.g. one that registers files
written outside Hudi for the first time,
+ // only has the schema of the current commit.
+ tableSchema = tryResolveSchemaForTable(dataMetaClient)
+ .orElseGet(() ->
Option.ofNullable(commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY))
Review Comment:
Fixed in 9085866. The fallback lives in `resolveTableSchema`, outside the
`try` that wraps `tryResolveSchemaForTable`, treats an empty `SCHEMA_KEY` as
absent, and the test uses `addMetadata(SCHEMA_KEY, "")` and asserts the full
message.
--
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]
