vinishjail97 commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3965697874
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/record/TestRecordIndexer.java:
##########
@@ -227,6 +238,82 @@ void
testBuildUpdateWithNonEmptyCommitMetadataProducesPartitionEntry() {
assertEquals(fileID, location.getFileId());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void testBuildUpdateDeletesRecordsOfFileGroupsReplacedByExternalWriter() {
+ // files written outside Hudi are registered through replace commits with
an unknown operation type. The records of the
+ // replaced file groups leave the index unless the same key is written
again in the commit.
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(getDefaultStorageConf());
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
Review Comment:
Done in b10d8167: `mockMetaClientForUpdate` holds the shared mocks, and
`StorageConfiguration` is imported.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java:
##########
@@ -165,12 +185,21 @@ private static Set<Pair<String, Long>>
filterParquetRowKeys(HoodieStorage storag
AvroReadSupport.setAvroReadSchema(conf, readSchema.toAvroSchema());
AvroReadSupport.setRequestedProjection(conf, readSchema.toAvroSchema());
Set<Pair<String, Long>> rowKeys = new HashSet<>();
+ Option<String> relativeFilePath = basePath.map(path ->
FSUtils.getRelativePartitionPath(path, convertToStoragePath(filePath)));
long rowPosition = 0;
try (ParquetReader reader =
AvroParquetReader.builder(filePath).withConf(conf).build()) {
Object obj = reader.read();
while (obj != null) {
if (obj instanceof GenericRecord) {
- String recordKey = ((GenericRecord)
obj).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
+ Object recordKeyValue = ((GenericRecord)
obj).get(HoodieRecord.RECORD_KEY_METADATA_FIELD);
+ String recordKey;
+ if (recordKeyValue != null) {
+ recordKey = recordKeyValue.toString();
+ } else {
+ ValidationUtils.checkArgument(relativeFilePath.isPresent(),
Review Comment:
Done in b10d8167. With the table-level gate a file is read either fully
positional or fully keyed, so the per-row branch is gone. A Hudi-written file
with a null key fails with "Record key is missing in row N of <file>" instead
of an NPE. Same shape in the SI generator.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.ValueSource;
+
+import java.io.IOException;
+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 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 files carry no record key, so every row is keyed by the file path
relative to the table and the row position.
+ */
+public class TestExternalFileRecordAndSecondaryIndex extends
HoodieClientTestBase {
+
+ private static final String ID_FIELD = "id";
+ private static final String NAME_FIELD = "name";
+ 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))));
+
+ @ParameterizedTest
+ @ValueSource(strings = {"americas/brazil", ""})
+ public void testRecordAndSecondaryIndexForExternalFiles(String
partitionPath) throws Exception {
+ // tables registered from other formats carry neither meta fields nor
record key fields
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty(HoodieTableConfig.POPULATE_META_FIELDS.key(),
"false");
+ tableProperties.setProperty(HoodieTableConfig.RECORDKEY_FIELDS.key(), "");
+ initMetaClient(tableProperties);
+ HoodieWriteConfig writeConfig = 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(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withMetadataIndexColumnStats(false)
+ .withEnableGlobalRecordLevelIndex(true)
+ .withSecondaryIndexEnabled(true)
+ .withSecondaryIndexForColumn(NAME_FIELD)
+ .build())
+ .build();
+ writeClient = getHoodieWriteClient(writeConfig);
+ writeClient.setOperationType(WriteOperationType.UNKNOWN);
+
+ // first commit registers one file with three rows
+ String fileName1 = "file_1.parquet";
+ List<Pair<Integer, String>> rows1 = Arrays.asList(Pair.of(1, "alice"),
Pair.of(2, "bob"), Pair.of(3, "alice"));
+ commitExternalFile(partitionPath, fileName1, rows1,
Collections.emptyMap());
+
+ String relativePath1 = relativeFilePath(partitionPath, fileName1);
+ HoodieBackedTableMetadata tableMetadata = new HoodieBackedTableMetadata(
+ context, metaClient.getStorage(), writeConfig.getMetadataConfig(),
writeConfig.getBasePath(), true);
+ Map<String, HoodieRecordGlobalLocation> recordIndex =
readRecordIndex(tableMetadata, generatedKeys(relativePath1, 3));
+ assertEquals(3, recordIndex.size());
+ recordIndex.values().forEach(location -> {
+ assertEquals(partitionPath, location.getPartitionPath());
+ assertEquals(fileName1, location.getFileId());
+ });
+ String secondaryIndexPartition = secondaryIndexPartition();
+ assertEquals(new HashMap<String, Set<String>>() {
Review Comment:
Done in b10d8167.
##########
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:
Done in b10d8167: passthrough case plus the nested `bucket-0/subdir` prefix
input.
##########
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:
Not supported for key-less external files, and I would keep it out of this
PR. `SecondaryIndexer.buildUpdate` rejects every insert-overwrite type
outright, so SI can never work under that label, and an external writer's
replace commit adds and removes specific files rather than overwriting a
partition, so `UNKNOWN` is the honest type (XTable sets it).
`TestExternalPathHandling` uses `INSERT_OVERWRITE` with column stats only. With
the table-level gate, a key-less file under `INSERT_OVERWRITE` now fails with
an explicit "Record key is missing" message instead of the NPE. I can follow up
on the reader if you want that path supported.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.ValueSource;
+
+import java.io.IOException;
+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 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 files carry no record key, so every row is keyed by the file path
relative to the table and the row position.
+ */
+public class TestExternalFileRecordAndSecondaryIndex extends
HoodieClientTestBase {
+
+ private static final String ID_FIELD = "id";
+ private static final String NAME_FIELD = "name";
+ 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))));
+
+ @ParameterizedTest
+ @ValueSource(strings = {"americas/brazil", ""})
+ public void testRecordAndSecondaryIndexForExternalFiles(String
partitionPath) throws Exception {
+ // tables registered from other formats carry neither meta fields nor
record key fields
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty(HoodieTableConfig.POPULATE_META_FIELDS.key(),
"false");
+ tableProperties.setProperty(HoodieTableConfig.RECORDKEY_FIELDS.key(), "");
+ initMetaClient(tableProperties);
+ HoodieWriteConfig writeConfig = 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(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withMetadataIndexColumnStats(false)
+ .withEnableGlobalRecordLevelIndex(true)
Review Comment:
Done in b10d8167: one parameterization runs with
`withEnableRecordLevelIndex(true)`, so the partitioned delete payload is
exercised.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -319,17 +380,32 @@ public boolean hasNext() {
while (recordIterator.hasNext()) {
T record = recordIterator.next();
Object secondaryKey =
readerContext.getRecordContext().getValue(record, requestedSchema,
secondaryKeyField);
- nextValidRecord = Pair.of(
- readerContext.getRecordContext().getRecordKey(record,
requestedSchema),
- secondaryKey == null ? null : secondaryKey.toString()
- );
+ nextValidRecord = Pair.of(getRecordKey(record), secondaryKey == null
? null : secondaryKey.toString());
+ rowPosition++;
return true;
}
// If no valid records are found
return false;
}
+ private String getRecordKey(T record) {
+ Object recordKey;
+ if (hasRecordKeyMetaField) {
+ recordKey = readerContext.getRecordContext().getValue(record,
requestedSchema, RECORD_KEY_METADATA_FIELD);
+ } else if (hasRecordKeyFields) {
+ recordKey = readerContext.getRecordContext().getRecordKey(record,
requestedSchema);
+ } else {
+ recordKey = null;
Review Comment:
Done in b10d8167. Keys are generated only when
`HoodieTableConfig.hasRecordKey()` is false, i.e. the record key meta column is
not populated and no record key fields are configured. I gated on the table
config rather than the marker because `HoodieBaseFile` strips the marker at
construction (`maybeHandleExternallyGeneratedFileName`), so replaced file
groups and bootstrap slices cannot see it. Selective `MetaFieldsMode` tables
have key fields, so they keep failing loudly on a null key. Applied to RLI
(write stats, replaced files, initialize) and SI (update, initialize).
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -197,6 +207,12 @@ public static <T> HoodieData<HoodieRecord>
convertWriteStatsToSecondaryIndexReco
return records.iterator();
});
+ if (commitMetadata instanceof HoodieReplaceCommitMetadata) {
Review Comment:
Done in b10d8167: `getSecondaryIndexUpdates` no longer returns early for a
replace commit with an unknown operation type and replaced file ids, even
without write stats. The third commit in
`testIndexesAreBuiltFromRegisteredFilesWhenEnabledLater` drops a file only and
asserts the SI and RLI entries disappear.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -319,17 +380,32 @@ public boolean hasNext() {
while (recordIterator.hasNext()) {
T record = recordIterator.next();
Object secondaryKey =
readerContext.getRecordContext().getValue(record, requestedSchema,
secondaryKeyField);
- nextValidRecord = Pair.of(
- readerContext.getRecordContext().getRecordKey(record,
requestedSchema),
- secondaryKey == null ? null : secondaryKey.toString()
- );
+ nextValidRecord = Pair.of(getRecordKey(record), secondaryKey == null
? null : secondaryKey.toString());
+ rowPosition++;
Review Comment:
Done in b10d8167: the generator throws when a key-less table's slice has log
files or no base file. The RLI initialize path has the same check.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/BaseFileRecordParsingUtils.java:
##########
@@ -167,10 +171,38 @@ public static Map<RecordStatus, List<String>>
getRecordKeyStatuses(String basePa
}
}
+ /**
+ * Generates RLI Metadata delete records for every record key in the given
base file.
+ * Used when a file group is replaced by a commit that does not rewrite its
records, for example a replace commit
+ * that registers files written outside Hudi.
+ *
+ * @param basePath base path of the table.
+ * @param partition partition of the base file.
+ * @param dataFilePath path of the base file on storage.
+ * @param storage instance of {@link HoodieStorage}.
+ * @param isPartitionedRLI whether the record index is partitioned.
+ * @return Iterator of delete {@link HoodieRecord}s for RLI Metadata
partition.
+ */
+ public static Iterator<HoodieRecord>
generateRLIMetadataHoodieRecordsForReplacedBaseFile(String basePath,
+
String partition,
+
StoragePath dataFilePath,
+
HoodieStorage storage,
+
boolean isPartitionedRLI) {
+ return getRecordKeysFromBaseFile(storage, basePath, dataFilePath).stream()
+ .map(recordKey ->
HoodieMetadataPayload.createRecordIndexDelete(recordKey, partition,
isPartitionedRLI))
+ .iterator();
+ }
+
private static Set<String> getRecordKeysFromBaseFile(HoodieStorage storage,
String basePath, String partition, String fileName) {
- StoragePath dataFilePath = new StoragePath(basePath,
StringUtils.isNullOrEmpty(partition) ? fileName : (partition +
StoragePath.SEPARATOR) + fileName);
+ // a file written outside Hudi is recorded with an external file marker
that is not part of the name on storage.
+ String filePathInPartition =
ExternalFilePathUtil.getFilePathInPartition(fileName);
+ StoragePath dataFilePath = new StoragePath(basePath,
StringUtils.isNullOrEmpty(partition) ? filePathInPartition : (partition +
StoragePath.SEPARATOR) + filePathInPartition);
+ return getRecordKeysFromBaseFile(storage, basePath, dataFilePath);
+ }
+
+ private static Set<String> getRecordKeysFromBaseFile(HoodieStorage storage,
String basePath, StoragePath dataFilePath) {
FileFormatUtils fileFormatUtils =
HoodieIOFactory.getIOFactory(storage).getFileFormatUtils(HoodieFileFormat.PARQUET);
- return fileFormatUtils.readRowKeys(storage, dataFilePath);
+ return fileFormatUtils.readRowKeys(storage, dataFilePath, new
StoragePath(basePath));
Review Comment:
Addressed in b10d8167 through the table-level gate, see the thread above.
--
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]
