voonhous commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3965129497
##########
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);
+ } else if (commitMetadata instanceof HoodieReplaceCommitMetadata &&
operationType != WriteOperationType.CLUSTER) {
+ // a replace commit that is neither a table service nor an overwrite,
e.g. files written outside Hudi being
+ // registered in the table. The replaced file groups are dropped without
their records being rewritten under
+ // the same key, so the records of the replaced base files are deleted
from RLI unless this commit wrote them again.
+ return
getRecordIndexReplacedFileGroupRecords((HoodieReplaceCommitMetadata)
commitMetadata, fsView)
+ .mapToPair(r -> Pair.of(r.getKey(), r))
+ .leftOuterJoin(updatesFromWriteStatuses.mapToPair(r ->
Pair.of(r.getKey(), r)))
+ .values()
+ .filter(p -> !p.getRight().isPresent())
+ .map(Pair::getLeft);
} else {
return engineContext.emptyHoodieData();
}
}
+ /**
+ * Reads the record keys of the latest base file of every file group
replaced by the given commit and
+ * returns a delete record for each of them.
+ */
+ private HoodieData<HoodieRecord>
getRecordIndexReplacedFileGroupRecords(HoodieReplaceCommitMetadata
replaceCommitMetadata, Lazy<HoodieTableFileSystemView> fsView) {
Review Comment:
**major:** This covers the incremental path, but RLI initialize still cannot
run over registered external files. `readRecordKeysFromFileSliceSnapshot` (line
268) projects the key from `populateMetaFields ? recordKeySchema :
projectSchema(dataSchema, recordKeyFields)`, which is empty for this table
config, and line 288 hardcodes `fileIdEncoding = 0`, so
`HoodieMetadataPayload:703` throws `Invalid UUID or index:
fileID=file_1.parquet`. SI initialize goes through the fixed generator, so the
two indexes diverge on bootstrap. The functional test never reaches this
because init returns empty when there is no completed commit. Could the init
path get the positional fallback plus `getWritesFileIdEncoding()`, with a test
that registers files first and enables RLI+SI afterwards?
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/secondary/TestSecondaryIndexer.java:
##########
@@ -171,6 +173,34 @@ void testBuildUpdateForCompactReturnsEmpty() {
commitMetadata)).isEmpty());
}
+ @Test
+ void testBuildUpdateForReplaceCommitFromExternalWriterIsNotRejected() {
+ // files written outside Hudi are registered through replace commits with
an unknown operation type
+ HoodieEngineContext engineContext = mock(HoodieEngineContext.class);
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ HoodieIndexMetadata indexMetadata = mock(HoodieIndexMetadata.class);
+ HoodieIndexDefinition indexDefinition = mock(HoodieIndexDefinition.class);
+
+
when(metaClient.getIndexMetadata()).thenReturn(org.apache.hudi.common.util.Option.of(indexMetadata));
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+
when(tableConfig.getMetadataPartitions()).thenReturn(Collections.emptySet());
+
when(indexMetadata.getIndexDefinitions()).thenReturn(Collections.singletonMap("secondary_index_idx",
indexDefinition));
+ when(indexDefinition.getIndexName()).thenReturn("secondary_index_idx");
+
+ HoodieReplaceCommitMetadata commitMetadata = new
HoodieReplaceCommitMetadata();
+ commitMetadata.setOperationType(WriteOperationType.UNKNOWN);
+ commitMetadata.addReplaceFileId("p1", "file_1.parquet");
+
+ SecondaryIndexer indexer = new SecondaryIndexer(engineContext,
writeConfig, metaClient);
+ assertTrue(indexer.buildUpdate(IndexUpdateContext.of(
Review Comment:
**major:** This test passes with the production change reverted. `UNKNOWN`
is non-null, so the added `operationType != null` guard is never the deciding
term, and `getMetadataPartitions()` is stubbed to an empty set, so
`buildUpdate` returns an empty list before the loop body runs, whatever the
branch does. Could it leave `operationType` unset (null, which is what a client
that never calls `setOperationType` produces) and stub a `secondary_index_*`
partition, asserting one `IndexPartitionAndRecords` instead of empty?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/secondary/SecondaryIndexer.java:
##########
@@ -115,7 +115,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()) {
Review Comment:
**minor:** Not blocking. The same value is dereferenced unguarded in
`PartitionStatsIndexer.buildUpdate:117`
(`getOperationType().equals(DELETE_PARTITION)`), so a null operation type NPEs
there whenever column stats is on, regardless of this guard. XTable itself sets
`UNKNOWN`, so this has the same defensive scope as this line. Should that call
get the same null tolerance, or should `HoodieCommitMetadata.setOperationType`
reject null so it fails in one place?
##########
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);
+ } else if (commitMetadata instanceof HoodieReplaceCommitMetadata &&
operationType != WriteOperationType.CLUSTER) {
Review Comment:
**minor:** Not blocking. This is an exclusion list of one, so `null`,
`UNKNOWN`, and any replace operation type added later all inherit "delete every
record of every replaced file group" plus a full read of those files. Today
only `null` / `UNKNOWN` reach it (bucket rescale is recorded as
`INSERT_OVERWRITE`, `DatasetBulkInsertOverwriteCommitActionExecutor:104`).
Could the condition be positive, so a future operation type has to opt in?
```suggestion
} else if (commitMetadata instanceof HoodieReplaceCommitMetadata
&& (operationType == null || operationType ==
WriteOperationType.UNKNOWN)) {
```
##########
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:
**minor:** Not blocking. These mock lines are the setup of
`testBuildUpdateWithNonEmptyCommitMetadataProducesPartitionEntry` (lines
177-196) copied verbatim, inline
`(org.apache.hudi.storage.StorageConfiguration)` cast included. Could the block
move to a private fixture helper, with `StorageConfiguration` imported so the
cast reads normally in both places?
##########
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:
**minor:** Not blocking. This `checkArgument` sits inside the per-row loop
and its message is a string concat, so on the common path for an external file
(every row null-keyed) it builds the message once per row. `relativeFilePath`
is already computed above the loop. Could the presence check move up there too,
so the loop only calls `relativeFilePath.get()`? Same pattern at
`SecondaryIndexRecordGenerationUtils:404`.
##########
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:
**minor:** Not blocking, pre-existing. These three branches read keys
through `readRecordKeysFromBaseFiles`, i.e.
`HoodieAvroParquetReader.getRecordKeyIterator`, whose line 231 is
`.get(RECORD_KEY_METADATA_FIELD).toString()` with no null guard, the pattern
this PR guards in `ParquetUtils`. It only matters if an external writer uses
`INSERT_OVERWRITE`; XTable uses `UNKNOWN`, but Hudi's own
`TestExternalPathHandling:115` does. Is `INSERT_OVERWRITE` meant to be
supported for external files, and if so should that reader get the same
fallback?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -287,6 +341,12 @@ private static <T> ClosableIterator<Pair<String, String>>
createSecondaryIndexRe
boolean allowInflightInstants) throws IOException {
String secondaryKeyField = indexDefinition.getSourceFieldsKey();
HoodieSchema requestedSchema =
getRequestedSchemaForSecondaryIndex(metaClient, tableSchema, secondaryKeyField);
+ // Files written outside Hudi may carry neither the record key meta field
nor record key fields. Their rows are
+ // keyed by the file path relative to the table base path and the row
position, the same key the record index uses.
+ boolean hasRecordKeyMetaField =
tableSchema.getField(RECORD_KEY_METADATA_FIELD).isPresent();
+ boolean hasRecordKeyFields =
metaClient.getTableConfig().getRecordKeyFields().map(fields -> fields.length >
0).orElse(false);
+ Option<String> relativeFilePath = fileSlice.getBaseFile()
Review Comment:
**major:** This side uses `baseFile.getStoragePath()`, which keeps the
`fg%3D<prefix>` directory. The SI initialize path at line 315 rebuilds the path
as `getAbsoluteFilePath(basePath, partition, baseFile.getFileName())`;
`getFileName()` is the last path component, so `base/p/bucket-0/f.parquet`
becomes `base/p/f.parquet` and `readSchema` opens a missing file. #17788 fixed
this exact shape, and `TestExternalPathHandling` already parameterizes it as
`paimonExternal`. Could line 315 use `getStoragePath()` too, and could that
prefix generator be added to the functional coverage?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -107,7 +114,10 @@ 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(() ->
HoodieSchema.parse(commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY)));
Review Comment:
**minor:** Not blocking. `getMetadata(SCHEMA_KEY)` returns null when the key
is absent (or when `commitMetadata` is null, as in the two unit tests), so
`HoodieSchema.parse(null)` NPEs and surfaces as "Failed to get latest schema
for ...", pointing at the wrong cause. The fallback itself has no positive
test. Could it name the missing key explicitly, plus one test that enters this
branch?
```suggestion
.orElseGet(() ->
Option.ofNullable(commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY))
.map(HoodieSchema::parse)
.orElseThrow(() -> new HoodieException("No completed commit
and no " + HoodieCommitMetadata.SCHEMA_KEY + " in commit metadata")));
```
##########
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 {
Review Comment:
**minor:** Not blocking. `TestExternalPathHandling` (hudi-spark-client)
already drives this exact registration flow, parameterized over the flat and
`paimonExternal` (`bucket-0/` prefix) name shapes and over partitioned /
unpartitioned tables, with `createWriteStatus` / `getPath` helpers of the same
shape as `commitExternalFile` here. Could the RLI/SI assertions land there as a
second `@ParameterizedTest`, or at least borrow its `FileIdAndNameGenerator`,
so the prefix shape is covered end to end?
##########
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:
**nit:** Feel free to ignore. `getFilePathInPartition` is a one-line
delegate to `parseFileIdAndCommitTimeFromExternalFile`, whose legacy and prefix
inputs `testParseFileIdAndCommitTimeFromExternalFile_LegacyFormat` /
`_WithPrefix` already assert. Could this keep only the non-external passthrough
case and add the nested-prefix input that
`testGetFullPath_OfPartition_WithNestedPrefix_2Levels` covers for the sibling
method?
##########
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:
**nit:** Feel free to ignore. Double-brace initialization creates an
anonymous `HashMap` subclass that captures the test instance (here and at line
141), and line 213 spells out `java.util.stream.IntStream` inline. Could these
use a `mapOf(...)` helper next to the existing `setOf(...)`, plus an
`IntStream` import?
##########
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:
**nit:** Feel free to ignore. Every new test pins `isPartitionedRLI ==
false` (global RLI here, `eq(false)` in `TestRecordIndexer`), so the
`EmptyHoodieRecordPayloadWithPartition` delete variant that
`createRecordIndexDelete` produces for a partitioned RLI is never constructed.
Would one parameterization with `withEnableRecordLevelIndex(true)` be worth
adding?
--
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]
