This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 1f8e53a703fa test: Coverage clustering merge bootstrap (#19493)
1f8e53a703fa is described below
commit 1f8e53a703fa0d3adb1a9d33855c8e1b69bc8612
Author: voonhous <[email protected]>
AuthorDate: Tue Aug 4 19:30:43 2026 +0800
test: Coverage clustering merge bootstrap (#19493)
* test(client): cover HoodieWriteMergeHandle drop, ignore,
partition-mismatch and reclose paths
* test(spark): cover COW table compact guard, handleInsert, getUpdateHandle
and rollbackBootstrap
* test(spark): cover the ORC bootstrap metadata handler
* test(spark): cover stream-copy clustering plan and execution strategies
* test(spark): cover external-file clustering execution strategy and write
handle
* test(client): cover HoodieMergeHelper bootstrap and schema-evolution paths
* test(spark): address review nits in binary-copy clustering test
Use the diamond operator for the two raw HashSet constructions and fix
the dynmicBloomFilter typo, including the pre-existing occurrences in
testSupportBinaryStreamCopy so the file stays consistent.
---
.../io/TestSortedAndChangeLogMergeHandles.java | 113 +++++++
.../hudi/client/TestUpdateSchemaEvolution.java | 137 +++++++-
.../TestSparkStreamCopyClusteringPlanStrategy.java | 342 +++++++++++++++++++
...arkExternalFileClusteringExecutionStrategy.java | 363 +++++++++++++++++++++
.../bootstrap/TestOrcBootstrapMetadataHandler.java | 237 ++++++++++++++
.../commit/TestCopyOnWriteActionExecutor.java | 125 +++++++
.../apache/hudi/functional/TestBootstrapRead.java | 43 +++
...SparkBinaryCopyClusteringAndValidationMeta.java | 141 +++++++-
8 files changed, 1484 insertions(+), 17 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestSortedAndChangeLogMergeHandles.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestSortedAndChangeLogMergeHandles.java
index 4949a926e6ec..784e3c1edba9 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestSortedAndChangeLogMergeHandles.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestSortedAndChangeLogMergeHandles.java
@@ -61,7 +61,10 @@ import java.util.Map;
import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
@@ -157,6 +160,85 @@ public class TestSortedAndChangeLogMergeHandles {
}
}
+ @Test
+ public void testWriteUpdateRecordDropsIncomingRecordWhenMergeKeepsOldValue()
throws Exception {
+ HoodieWriteConfig config = config();
+ TestContext context = new TestContext(config);
+
+ try (MockedStatic<WriteMarkersFactory> markers =
mockStatic(WriteMarkersFactory.class);
+ MockedStatic<HoodieFileWriterFactory> writers =
mockStatic(HoodieFileWriterFactory.class)) {
+ context.stubWriters(markers, writers);
+ TestableWriteMergeHandle handle = new TestableWriteMergeHandle(config,
context.table, new HashMap<>());
+
+ // The merge kept the old value, so the combined record carries the old
payload instance.
+ GenericRecord oldData = mock(GenericRecord.class);
+ HoodieRecord oldRecord = record("u");
+ HoodieRecord combinedRecord = record("u");
+ when(oldRecord.getData()).thenReturn(oldData);
+ when(combinedRecord.getData()).thenReturn(oldData);
+
+ assertFalse(handle.writeUpdate(record("u"), oldRecord, combinedRecord));
+ assertEquals(Collections.emptyList(), handle.writtenKeys);
+ }
+ }
+
+ @Test
+ public void testWriteInsertRecordSkipsIgnoredRecord() throws Exception {
+ HoodieWriteConfig config = config();
+ TestContext context = new TestContext(config);
+
+ try (MockedStatic<WriteMarkersFactory> markers =
mockStatic(WriteMarkersFactory.class);
+ MockedStatic<HoodieFileWriterFactory> writers =
mockStatic(HoodieFileWriterFactory.class)) {
+ context.stubWriters(markers, writers);
+ TestableWriteMergeHandle handle = new TestableWriteMergeHandle(config,
context.table, new HashMap<>());
+
+ HoodieRecord ignored = record("i");
+ when(ignored.shouldIgnore(any(HoodieSchema.class),
any())).thenReturn(true);
+ handle.writeInsert(ignored);
+
+ assertEquals(Collections.emptyList(), handle.writtenKeys);
+ }
+ }
+
+ @Test
+ public void testWriteRecordMarksFailureWhenPartitionPathDoesNotMatch()
throws Exception {
+ HoodieWriteConfig config = config();
+ TestContext context = new TestContext(config);
+
+ try (MockedStatic<WriteMarkersFactory> markers =
mockStatic(WriteMarkersFactory.class);
+ MockedStatic<HoodieFileWriterFactory> writers =
mockStatic(HoodieFileWriterFactory.class)) {
+ context.stubWriters(markers, writers);
+ TestableWriteMergeHandle handle = new TestableWriteMergeHandle(config,
context.table, new HashMap<>());
+
+ HoodieRecord foreignRecord = record("f");
+ when(foreignRecord.getPartitionPath()).thenReturn("another-partition");
+ handle.writeInsert(foreignRecord);
+
+ assertEquals(Collections.emptyList(), handle.writtenKeys);
+ assertTrue(handle.status().hasErrors());
+ assertEquals(1, handle.status().getTotalErrorRecords());
+ }
+ }
+
+ @Test
+ public void testCloseIsIdempotent() throws Exception {
+ HoodieWriteConfig config = config();
+ TestContext context = new TestContext(config);
+
+ try (MockedStatic<WriteMarkersFactory> markers =
mockStatic(WriteMarkersFactory.class);
+ MockedStatic<HoodieFileWriterFactory> writers =
mockStatic(HoodieFileWriterFactory.class)) {
+ context.stubWriters(markers, writers);
+ TestableWriteMergeHandle handle = new TestableWriteMergeHandle(config,
context.table, new HashMap<>());
+
+ List<WriteStatus> first = handle.close();
+ List<WriteStatus> second = handle.close();
+
+ assertEquals(1, second.size());
+ assertSame(first.get(0), second.get(0));
+ verify(context.fileWriter, times(1)).close();
+ }
+ }
+
private static HoodieWriteConfig config() {
return HoodieWriteConfig.newBuilder()
.withPath("/tmp")
@@ -244,6 +326,37 @@ public class TestSortedAndChangeLogMergeHandles {
}
}
+ private static class TestableWriteMergeHandle extends HoodieWriteMergeHandle
{
+ private final List<String> writtenKeys = new ArrayList<>();
+
+ private TestableWriteMergeHandle(
+ HoodieWriteConfig config, HoodieTable table, Map<String, HoodieRecord>
records) {
+ super(config, "100", table, records, "partition", "file-1", null,
+ new LocalTaskContextSupplier(), Option.empty());
+ }
+
+ @Override
+ protected void writeToFile(
+ HoodieKey key, HoodieRecord record, HoodieSchema schema, Properties
props,
+ boolean shouldPreserveRecordMetadata) {
+ // These tests target the decisions taken before the record reaches the
writer.
+ writtenKeys.add(key.getRecordKey());
+ }
+
+ private void writeInsert(HoodieRecord record) throws IOException {
+ writeInsertRecord(record);
+ }
+
+ private boolean writeUpdate(
+ HoodieRecord newRecord, HoodieRecord oldRecord, HoodieRecord
combinedRecord) throws IOException {
+ return writeUpdateRecord(newRecord, oldRecord, combinedRecord,
writeSchema);
+ }
+
+ private WriteStatus status() {
+ return writeStatus;
+ }
+ }
+
private static class TestableChangeLogMergeHandle extends
HoodieMergeHandleWithChangeLog {
private TestableChangeLogMergeHandle(
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
index 8bfa54bbb765..4690ebb5805d 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
@@ -18,16 +18,28 @@
package org.apache.hudi.client;
+import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
import org.apache.hudi.common.model.HoodieAvroPayload;
import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.schema.internal.InternalSchema;
+import org.apache.hudi.common.schema.internal.action.TableChanges;
+import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter;
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
+import org.apache.hudi.common.schema.internal.utils.SerDeHelper;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
+import org.apache.hudi.common.testutils.FileCreateUtilsLegacy;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.testutils.InProcessTimeGenerator;
+import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.JsonUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -50,22 +62,27 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.function.Executable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.io.Serializable;
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.function.Function;
import java.util.stream.Collectors;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.COMMIT_METADATA_SER_DE;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME_GENERATOR;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.createSimpleRecord;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.extractPartitionFromTimeField;
import static
org.apache.hudi.common.testutils.SchemaTestUtil.getSchemaFromResource;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class TestUpdateSchemaEvolution extends HoodieSparkClientTestHarness
implements Serializable {
@@ -85,6 +102,17 @@ public class TestUpdateSchemaEvolution extends
HoodieSparkClientTestHarness impl
}
private WriteStatus prepareFirstRecordCommit(List<HoodieRecord>
insertRecords) throws IOException {
+ return prepareFirstRecordCommit(insertRecords, Option.empty(), false);
+ }
+
+ /**
+ * Writes the first commit. When {@code writeCommitMetadata} is set the
commit file holds the schema the
+ * records were written with, and {@code internalSchema}, if given, is
recorded as the latest schema of that
+ * commit. That is how {@code HoodieMergeHelper} resolves the internal
schema of the base file being merged.
+ */
+ private WriteStatus prepareFirstRecordCommit(List<HoodieRecord>
insertRecords,
+ Option<InternalSchema>
internalSchema,
+ boolean writeCommitMetadata)
throws IOException {
// Create a bunch of records with an old version of schema
final HoodieWriteConfig config =
makeHoodieClientConfig("/exampleSchema.avsc");
final HoodieSparkTable table = HoodieSparkTable.create(config, context);
@@ -99,9 +127,18 @@ public class TestUpdateSchemaEvolution extends
HoodieSparkClientTestHarness impl
return createHandle.close().get(0);
}).collect();
- final Path commitFile = new Path(config.getBasePath() +
"/.hoodie/timeline/"
- + INSTANT_FILE_NAME_GENERATOR.makeCommitFileName("100" + "_" +
InProcessTimeGenerator.createNewInstantTime()));
- HadoopFSUtils.getFs(basePath,
HoodieTestUtils.getDefaultStorageConf()).create(commitFile);
+ if (writeCommitMetadata) {
+ HoodieCommitMetadata commitMetadata =
CommitUtils.buildMetadata(Collections.emptyList(), Collections.emptyMap(),
+ Option.empty(), WriteOperationType.INSERT, config.getSchema(),
HoodieTimeline.COMMIT_ACTION);
+ if (internalSchema.isPresent()) {
+ commitMetadata.addMetadata(SerDeHelper.LATEST_SCHEMA,
SerDeHelper.toJson(internalSchema.get()));
+ }
+ FileCreateUtilsLegacy.createCommit(COMMIT_METADATA_SER_DE, basePath,
"100", Option.of(commitMetadata));
+ } else {
+ final Path commitFile = new Path(config.getBasePath() +
"/.hoodie/timeline/"
+ + INSTANT_FILE_NAME_GENERATOR.makeCommitFileName("100" + "_" +
InProcessTimeGenerator.createNewInstantTime()));
+ HadoopFSUtils.getFs(basePath,
HoodieTestUtils.getDefaultStorageConf()).create(commitFile);
+ }
return statuses.get(0);
}
@@ -230,11 +267,101 @@ public class TestUpdateSchemaEvolution extends
HoodieSparkClientTestHarness impl
assertSchemaEvolutionOnUpdateResult(insertResult, table, updateRecords,
assertMsg, true, ParquetDecodingException.class);
}
+ /**
+ * When the write config carries an internal (schema-on-read) schema, {@code
HoodieMergeHelper} reconciles it
+ * against the internal schema of the base file and rewrites every record
read out of that file. Here the
+ * column {@code number} was renamed, so the values have to be carried over
to the renamed column.
+ */
+ @Test
+ public void testMergeHelperRewritesRecordsOnRenamedColumn() throws Exception
{
+ HoodieSchema fileSchema =
HoodieSchemaUtils.addMetadataFields(getSchemaFromResource(getClass(),
"/exampleSchema.avsc"));
+ InternalSchema fileInternalSchema =
InternalSchemaConverter.convert(fileSchema).setSchemaId(100L);
+ WriteStatus insertResult =
+ prepareFirstRecordCommit(generateMultipleRecordsForExampleSchema(),
Option.of(fileInternalSchema), true);
+
+ // rename `number` to `numberx`, which is what `alter table rename column`
leaves on the write config
+ TableChanges.ColumnUpdateChange renameChange =
TableChanges.ColumnUpdateChange.get(fileInternalSchema)
+ .renameColumn("number", "numberx");
+ InternalSchema querySchema =
SchemaChangeUtils.applyTableChanges2Schema(fileInternalSchema,
renameChange).setSchemaId(101L);
+ HoodieSchema renamedSchema =
+
HoodieSchemaUtils.removeMetadataFields(InternalSchemaConverter.convert(querySchema,
fileSchema.getFullName()));
+
+ HoodieWriteConfig config = makeHoodieClientConfig(renamedSchema,
Option.of(querySchema), false);
+ String recordStr =
"{\"_row_key\":\"8eb5b87a-1feh-4edd-87b4-6ec96dc405a0\","
+ + "\"time\":\"2016-01-31T03:16:41.415Z\",\"numberx\":34}";
+ List<GenericRecord> mergedRecords = runMerge(config, insertResult,
buildUpdateRecords(recordStr, insertResult.getFileId(), config.getSchema()));
+
+ // the two records that were not updated keep the value of the column they
were written with
+ assertMergedNumbers(mergedRecords, "numberx");
+ }
+
+ /**
+ * The commit of the base file has no internal schema recorded, which is how
a table looks when schema on
+ * read is turned on after some inserts. With reconciliation enabled the
merge helper falls back to the
+ * table schema resolver to get one, without it there is nothing to
reconcile the query schema against.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testMergeHelperWithoutInternalSchemaOnCommit(boolean
reconcileSchema) throws Exception {
+ // the table schema resolver needs the commit to carry the schema the
records were written with
+ WriteStatus insertResult =
+ prepareFirstRecordCommit(generateMultipleRecordsForExampleSchema(),
Option.empty(), reconcileSchema);
+
+ HoodieSchema schema = getSchemaFromResource(getClass(),
"/exampleSchema.avsc");
+ InternalSchema querySchema =
InternalSchemaConverter.convert(HoodieSchemaUtils.addMetadataFields(schema)).setSchemaId(101L);
+ HoodieWriteConfig config = makeHoodieClientConfig(schema,
Option.of(querySchema), reconcileSchema);
+ String recordStr =
"{\"_row_key\":\"8eb5b87a-1feh-4edd-87b4-6ec96dc405a0\","
+ + "\"time\":\"2016-01-31T03:16:41.415Z\",\"number\":34}";
+ List<GenericRecord> mergedRecords = runMerge(config, insertResult,
buildUpdateRecords(recordStr, insertResult.getFileId(), config.getSchema()));
+
+ assertMergedNumbers(mergedRecords, "number");
+ }
+
+ private void assertMergedNumbers(List<GenericRecord> mergedRecords, String
numberColumn) {
+ Map<String, Integer> numberByKey = new HashMap<>();
+ for (GenericRecord record : mergedRecords) {
+ numberByKey.put(record.get("_row_key").toString(), (Integer)
record.get(numberColumn));
+ }
+ assertEquals(3, numberByKey.size());
+ assertEquals(34, numberByKey.get("8eb5b87a-1feh-4edd-87b4-6ec96dc405a0"));
+ assertEquals(100, numberByKey.get("8eb5b87b-1feu-4edd-87b4-6ec96dc405a0"));
+ assertEquals(15, numberByKey.get("8eb5b87c-1fej-4edd-87b4-6ec96dc405a0"));
+ }
+
+ /**
+ * Merges {@code updateRecords} into the base file written by {@code
insertResult} through
+ * {@code HoodieMergeHelper} and returns the records of the file it produced.
+ */
+ private List<GenericRecord> runMerge(HoodieWriteConfig config, WriteStatus
insertResult,
+ List<HoodieRecord> updateRecords) {
+ HoodieSparkTable table = HoodieSparkTable.create(config, context);
+ List<String> mergedFilePaths = jsc.parallelize(Arrays.asList(1)).map(x -> {
+ HoodieWriteMergeHandle mergeHandle = new HoodieWriteMergeHandle(config,
"101", table,
+ updateRecords.iterator(), updateRecords.get(0).getPartitionPath(),
insertResult.getFileId(), supplier, Option.empty());
+ // `doMerge` is the only entry point into HoodieMergeHelper: it reads
the base file and feeds the handle
+ mergeHandle.doMerge();
+ return ((WriteStatus) mergeHandle.close().get(0)).getStat().getPath();
+ }).collect();
+
+ return HoodieIOFactory.getIOFactory(table.getStorage())
+ .getFileFormatUtils(table.getBaseFileFormat())
+ .readAvroRecords(table.getStorage(), new
StoragePath(config.getBasePath() + "/" + mergedFilePaths.get(0)));
+ }
+
private HoodieWriteConfig makeHoodieClientConfig(String name) {
- HoodieSchema schema = getSchemaFromResource(getClass(), name);
- return HoodieWriteConfig.newBuilder().withPath(basePath)
+ return makeHoodieClientConfig(getSchemaFromResource(getClass(), name),
Option.empty(), false);
+ }
+
+ private HoodieWriteConfig makeHoodieClientConfig(HoodieSchema schema,
Option<InternalSchema> internalSchema,
+ boolean reconcileSchema) {
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath(basePath)
.withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
.withRemoteServerPort(timelineServicePort).build())
+
.withProps(Collections.singletonMap(HoodieCommonConfig.RECONCILE_SCHEMA.key(),
String.valueOf(reconcileSchema)))
.withSchema(schema.toString()).build();
+ if (internalSchema.isPresent()) {
+ config.setInternalSchemaString(SerDeHelper.toJson(internalSchema.get()));
+ }
+ return config;
}
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/plan/strategy/TestSparkStreamCopyClusteringPlanStrategy.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/plan/strategy/TestSparkStreamCopyClusteringPlanStrategy.java
new file mode 100644
index 000000000000..01366ace20a7
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/plan/strategy/TestSparkStreamCopyClusteringPlanStrategy.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.clustering.plan.strategy;
+
+import org.apache.hudi.avro.HoodieAvroWriteSupport;
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.common.config.HoodieParquetConfig;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Lazy;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.io.storage.hadoop.HoodieAvroParquetWriter;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieSparkCopyOnWriteTable;
+
+import org.apache.avro.Schema;
+import org.apache.parquet.avro.AvroSchemaConverter;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.schema.MessageType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
+
+/**
+ * Tests the schema aware grouping of {@link
SparkStreamCopyClusteringPlanStrategy}. Files are written for real
+ * because the grouping keys off the parquet schema hash read back from
storage.
+ */
+public class TestSparkStreamCopyClusteringPlanStrategy {
+
+ private static final String PARTITION_PATH = "p0";
+ private static final String SCHEMA_ONE_FIELD =
"{\"type\":\"record\",\"name\":\"triprec\",\"fields\":["
+ + "{\"name\":\"_row_key\",\"type\":\"string\"}]}";
+ private static final String SCHEMA_TWO_FIELDS =
"{\"type\":\"record\",\"name\":\"triprec\",\"fields\":["
+ +
"{\"name\":\"_row_key\",\"type\":\"string\"},{\"name\":\"rider\",\"type\":\"string\"}]}";
+
+ @TempDir
+ Path tempDir;
+
+ private HoodieSparkCopyOnWriteTable table;
+ private HoodieLocalEngineContext context;
+
+ @BeforeEach
+ public void setUp() {
+ table = Mockito.mock(HoodieSparkCopyOnWriteTable.class);
+ context = new
HoodieLocalEngineContext(HoodieTestUtils.getDefaultStorageConf());
+ HoodieStorage storage =
HoodieTestUtils.getStorage(tempDir.toAbsolutePath().toString());
+ Mockito.when(table.getStorage()).thenReturn(storage);
+ }
+
+ @Test
+ public void testSchemaAwareGroupingSplitsOnSchemaMismatch() throws
IOException {
+ SparkStreamCopyClusteringPlanStrategy planStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context,
defaultConfigBuilder().build());
+
+ List<FileSlice> fileSlices = new ArrayList<>();
+ fileSlices.add(createFileSlice(400, writeParquetFile("001",
SCHEMA_ONE_FIELD)));
+ fileSlices.add(createFileSlice(300, writeParquetFile("002",
SCHEMA_TWO_FIELDS)));
+
+ Pair<Stream<HoodieClusteringGroup>, Boolean> result =
+ planStrategy.buildClusteringGroupsForPartition(PARTITION_PATH,
fileSlices);
+ List<HoodieClusteringGroup> clusteringGroups = collect(result);
+
+ // Both slices fit in one group size wise, but the schemas differ so a
group break is forced.
+ Assertions.assertEquals(2, clusteringGroups.size());
+ Assertions.assertEquals(1, clusteringGroups.get(0).getSlices().size());
+ Assertions.assertEquals(1, clusteringGroups.get(1).getSlices().size());
+ Assertions.assertFalse(result.getRight());
+ }
+
+ @Test
+ public void testSizeOnlyGroupingWhenSchemaEvolutionEnabled() throws
IOException {
+ HoodieWriteConfig config = defaultConfigBuilder()
+ .withClusteringConfig(clusteringConfigBuilder()
+ .withFileStitchingBinaryCopySchemaEvolutionEnabled(true)
+ .build())
+ .build();
+ SparkStreamCopyClusteringPlanStrategy planStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context, config);
+
+ List<FileSlice> fileSlices = new ArrayList<>();
+ fileSlices.add(createFileSlice(400, writeParquetFile("001",
SCHEMA_ONE_FIELD)));
+ fileSlices.add(createFileSlice(300, writeParquetFile("002",
SCHEMA_TWO_FIELDS)));
+
+ List<HoodieClusteringGroup> clusteringGroups =
+ collect(planStrategy.buildClusteringGroupsForPartition(PARTITION_PATH,
fileSlices));
+
+ // Schema evolution enabled falls back to the parent size only grouping,
so the schema change is ignored.
+ Assertions.assertEquals(1, clusteringGroups.size());
+ Assertions.assertEquals(2, clusteringGroups.get(0).getSlices().size());
+ }
+
+ @Test
+ public void testMaxNumGroupsReachedMarksPartialScheduling() throws
IOException {
+ HoodieWriteConfig config = defaultConfigBuilder()
+ .withClusteringConfig(clusteringConfigBuilder()
+ .withClusteringMaxBytesInGroup(500)
+ .withClusteringMaxNumGroups(1)
+ .build())
+ .build();
+ SparkStreamCopyClusteringPlanStrategy planStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context, config);
+
+ List<FileSlice> fileSlices = new ArrayList<>();
+ fileSlices.add(createFileSlice(400, writeParquetFile("001",
SCHEMA_ONE_FIELD)));
+ fileSlices.add(createFileSlice(300, writeParquetFile("002",
SCHEMA_ONE_FIELD)));
+ fileSlices.add(createFileSlice(200, writeParquetFile("003",
SCHEMA_ONE_FIELD)));
+
+ Pair<Stream<HoodieClusteringGroup>, Boolean> result =
+ planStrategy.buildClusteringGroupsForPartition(PARTITION_PATH,
fileSlices);
+ List<HoodieClusteringGroup> clusteringGroups = collect(result);
+
+ // The first group closes at 400 bytes which already hits the max group
count, the rest are left behind.
+ Assertions.assertEquals(1, clusteringGroups.size());
+ Assertions.assertEquals(1, clusteringGroups.get(0).getSlices().size());
+ Assertions.assertTrue(result.getRight(), "Remaining slices were not
scheduled, expecting partial scheduling");
+ }
+
+ @Test
+ public void testTrailingGroupHonoursSingleGroupClusteringConfig() throws
IOException {
+ String filePath = writeParquetFile("001", SCHEMA_ONE_FIELD);
+
+ SparkStreamCopyClusteringPlanStrategy enabledStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context,
defaultConfigBuilder().build());
+ List<HoodieClusteringGroup> enabledGroups =
collect(enabledStrategy.buildClusteringGroupsForPartition(
+ PARTITION_PATH, Collections.singletonList(createFileSlice(200,
filePath))));
+ Assertions.assertEquals(1, enabledGroups.size());
+ Assertions.assertEquals(1, enabledGroups.get(0).getSlices().size());
+ Assertions.assertEquals(1, enabledGroups.get(0).getNumOutputFileGroups());
+
+ HoodieWriteConfig disabledConfig = defaultConfigBuilder()
+ .withClusteringConfig(clusteringConfigBuilder()
+ .withSingleGroupClusteringEnabled(false)
+ .build())
+ .build();
+ SparkStreamCopyClusteringPlanStrategy disabledStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context,
disabledConfig);
+ List<HoodieClusteringGroup> disabledGroups =
collect(disabledStrategy.buildClusteringGroupsForPartition(
+ PARTITION_PATH, Collections.singletonList(createFileSlice(200,
filePath))));
+ Assertions.assertEquals(0, disabledGroups.size());
+ }
+
+ @Test
+ public void testSchemaHashFallsBackToZeroForMissingFiles() throws
IOException {
+ HoodieWriteConfig config = defaultConfigBuilder()
+ .withClusteringConfig(clusteringConfigBuilder()
+ // large enough that a slice without a base file (sized as one
parquet max file size) still fits
+ .withClusteringMaxBytesInGroup(1024 * 1024 * 1024L)
+ .withClusteringTargetFileMaxBytes(1024 * 1024 * 1024L)
+ .build())
+ .build();
+ SparkStreamCopyClusteringPlanStrategy planStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context, config);
+
+ List<FileSlice> fileSlices = new ArrayList<>();
+ // no base file at all, sorts first since it is sized as one parquet max
file size
+ fileSlices.add(new FileSlice(PARTITION_PATH, "001",
FSUtils.createNewFileId(FSUtils.createNewFileIdPfx(), 0)));
+ fileSlices.add(createFileSlice(400, writeParquetFile("002",
SCHEMA_ONE_FIELD)));
+ fileSlices.add(createFileSlice(300, new StoragePath(
+ tempDir.toAbsolutePath().toString(),
+ FSUtils.makeBaseFileName("003", "1-0-1",
FSUtils.createNewFileId(FSUtils.createNewFileIdPfx(), 0),
".parquet")).toString()));
+
+ List<HoodieClusteringGroup> clusteringGroups =
+ collect(planStrategy.buildClusteringGroupsForPartition(PARTITION_PATH,
fileSlices));
+
+ // hash 0 (no base file), the real schema hash and hash 0 again
(unreadable file) - every neighbour mismatches
+ Assertions.assertEquals(3, clusteringGroups.size());
+ clusteringGroups.forEach(group -> Assertions.assertEquals(1,
group.getSlices().size()));
+ }
+
+ @Test
+ public void testSchemaHashFallsBackToZeroWhenStorageIsUnavailable() throws
IOException {
+ List<FileSlice> fileSlices = new ArrayList<>();
+ fileSlices.add(createFileSlice(400, writeParquetFile("001",
SCHEMA_ONE_FIELD)));
+ fileSlices.add(createFileSlice(300, writeParquetFile("002",
SCHEMA_TWO_FIELDS)));
+
+ Mockito.doThrow(new HoodieIOException("storage is
down")).when(table).getStorage();
+ SparkStreamCopyClusteringPlanStrategy planStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context,
defaultConfigBuilder().build());
+
+ List<HoodieClusteringGroup> clusteringGroups =
+ collect(planStrategy.buildClusteringGroupsForPartition(PARTITION_PATH,
fileSlices));
+
+ // both hashes fall back to 0, so the differing schemas no longer break
the group
+ Assertions.assertEquals(1, clusteringGroups.size());
+ Assertions.assertEquals(2, clusteringGroups.get(0).getSlices().size());
+ }
+
+ @Test
+ public void testGetStrategyParams() {
+ SparkStreamCopyClusteringPlanStrategy withoutSortColumns =
+ new SparkStreamCopyClusteringPlanStrategy(table, context,
defaultConfigBuilder().build());
+ Assertions.assertTrue(withoutSortColumns.getStrategyParams().isEmpty());
+
+ HoodieWriteConfig config = defaultConfigBuilder()
+ .withClusteringConfig(clusteringConfigBuilder()
+ .withClusteringSortColumns("col1,col2")
+ .build())
+ .build();
+ Map<String, String> params =
+ new SparkStreamCopyClusteringPlanStrategy(table, context,
config).getStrategyParams();
+ Assertions.assertEquals(1, params.size());
+ Assertions.assertEquals("col1,col2",
params.get(PLAN_STRATEGY_SORT_COLUMNS.key()));
+ }
+
+ @Test
+ public void testGenerateClusteringPlanOverridesExecutionStrategy() throws
IOException {
+ HoodieWriteConfig config = defaultConfigBuilder()
+ .withClusteringConfig(clusteringConfigBuilder()
+ .withClusteringSortColumns("col1")
+ .build())
+ .build();
+
+ FileSlice slice1 = createFileSlice(400, writeParquetFile("001",
SCHEMA_ONE_FIELD));
+ FileSlice slice2 = createFileSlice(300, writeParquetFile("002",
SCHEMA_ONE_FIELD));
+
+ HoodieTableMetaClient metaClient =
Mockito.mock(HoodieTableMetaClient.class);
+ Mockito.when(table.getMetaClient()).thenReturn(metaClient);
+ SyncableFileSystemView sliceView =
Mockito.mock(SyncableFileSystemView.class);
+ Mockito.when(table.getSliceView()).thenReturn(sliceView);
+
Mockito.when(sliceView.getPendingCompactionOperations()).thenAnswer(invocation
-> Stream.empty());
+
Mockito.when(sliceView.getPendingLogCompactionOperations()).thenAnswer(invocation
-> Stream.empty());
+
Mockito.when(sliceView.getFileGroupsInPendingClustering()).thenAnswer(invocation
-> Stream.empty());
+ Mockito.when(sliceView.getLatestFileSlicesStateless(PARTITION_PATH))
+ .thenReturn(Stream.of(slice1, slice2))
+ .thenAnswer(invocation -> Stream.empty());
+
+ SparkStreamCopyClusteringPlanStrategy planStrategy =
+ new SparkStreamCopyClusteringPlanStrategy(table, context, config);
+ Option<HoodieClusteringPlan> planOption =
+ planStrategy.generateClusteringPlan(null,
Lazy.eagerly(Collections.singletonList(PARTITION_PATH)));
+
+ Assertions.assertTrue(planOption.isPresent());
+ HoodieClusteringPlan plan = planOption.get();
+
Assertions.assertEquals(HoodieClusteringConfig.SPARK_STREAM_COPY_CLUSTERING_EXECUTION_STRATEGY,
+ plan.getStrategy().getStrategyClassName());
+ Assertions.assertEquals("col1",
plan.getStrategy().getStrategyParams().get(PLAN_STRATEGY_SORT_COLUMNS.key()));
+ Assertions.assertTrue(plan.getPreserveHoodieMetadata());
+ Assertions.assertEquals(1, plan.getInputGroups().size());
+ Assertions.assertEquals(2,
plan.getInputGroups().get(0).getSlices().size());
+
+ // nothing eligible on the second pass, the empty plan of the parent is
passed through untouched
+ Assertions.assertFalse(planStrategy.generateClusteringPlan(
+ null,
Lazy.eagerly(Collections.singletonList(PARTITION_PATH))).isPresent());
+ }
+
+ private static HoodieWriteConfig.Builder defaultConfigBuilder() {
+ return HoodieWriteConfig.newBuilder()
+ .withPath("")
+ .withClusteringConfig(clusteringConfigBuilder().build());
+ }
+
+ private static HoodieClusteringConfig.Builder clusteringConfigBuilder() {
+ return HoodieClusteringConfig.newBuilder()
+
.withClusteringPlanStrategyClass(HoodieClusteringConfig.SPARK_STREAM_COPY_CLUSTERING_PLAN_STRATEGY)
+ .withClusteringMaxBytesInGroup(2000)
+ .withClusteringTargetFileMaxBytes(1000)
+ .withClusteringPlanSmallFileLimit(1000);
+ }
+
+ private static List<HoodieClusteringGroup>
collect(Pair<Stream<HoodieClusteringGroup>, Boolean> result) {
+ return result.getLeft().collect(Collectors.toList());
+ }
+
+ private FileSlice createFileSlice(long baseFileSize, String filePath) {
+ HoodieBaseFile baseFile = new HoodieBaseFile(filePath);
+ baseFile.setFileSize(baseFileSize);
+ FileSlice fileSlice = new FileSlice(PARTITION_PATH,
baseFile.getCommitTime(), baseFile.getFileId());
+ fileSlice.setBaseFile(baseFile);
+ return fileSlice;
+ }
+
+ /**
+ * Writes an empty parquet file with the given schema, so that the strategy
has a real schema hash to read back.
+ */
+ private String writeParquetFile(String commitTime, String schemaStr) throws
IOException {
+ Schema avroSchema = new Schema.Parser().parse(schemaStr);
+ MessageType messageType = new AvroSchemaConverter().convert(avroSchema);
+ HoodieAvroWriteSupport writeSupport = new HoodieAvroWriteSupport(
+ messageType, HoodieSchema.fromAvroSchema(avroSchema), Option.empty(),
new Properties());
+ String fileName = FSUtils.makeBaseFileName(
+ commitTime, "1-0-1",
FSUtils.createNewFileId(FSUtils.createNewFileIdPfx(), 0), ".parquet");
+ StoragePath filePath = new
StoragePath(tempDir.resolve(fileName).toAbsolutePath().toString());
+ HoodieParquetConfig<HoodieAvroWriteSupport> parquetConfig = new
HoodieParquetConfig<>(
+ writeSupport,
+ CompressionCodecName.GZIP,
+ ParquetWriter.DEFAULT_BLOCK_SIZE,
+ ParquetWriter.DEFAULT_PAGE_SIZE,
+ 1024 * 1024 * 1024,
+ HoodieTestUtils.getDefaultStorageConf(),
+ 0.1,
+ true);
+ try (HoodieAvroParquetWriter writer =
+ new HoodieAvroParquetWriter(filePath, parquetConfig, commitTime,
new LocalTaskContextSupplier(), true)) {
+ // nothing to write, only the schema in the footer matters here
+ }
+ return filePath.toString();
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/run/strategy/TestSparkExternalFileClusteringExecutionStrategy.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/run/strategy/TestSparkExternalFileClusteringExecutionStrategy.java
new file mode 100644
index 000000000000..ca367b218b61
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/run/strategy/TestSparkExternalFileClusteringExecutionStrategy.java
@@ -0,0 +1,363 @@
+/*
+ * 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.clustering.run.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieSliceInfo;
+import org.apache.hudi.client.ExternalFileClusteringTestExecutionStrategy;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import
org.apache.hudi.client.clustering.plan.strategy.SparkSingleFileSortPlanStrategy;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ClusteringGroupInfo;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.IOType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ParquetUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieClusteringException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.WriteMarkersFactory;
+import org.apache.hudi.testutils.HoodieClientTestUtils;
+import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.sql.Row;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
+import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link SparkExternalFileClusteringExecutionStrategy} and the
+ * {@link org.apache.hudi.io.ExternalFileClusteringWriteHandle} it drives,
using the
+ * {@link ExternalFileClusteringTestExecutionStrategy} which transforms a file
by copying it as is.
+ */
+public class TestSparkExternalFileClusteringExecutionStrategy extends
HoodieSparkClientTestHarness {
+
+ private static final String PARTITION_PATH =
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+
+ private HoodieWriteConfig config;
+
+ @BeforeEach
+ public void setUp() throws IOException {
+ initPath();
+ initSparkContexts();
+ initTestDataGenerator();
+ initHoodieStorage();
+ Properties props = getPropertiesForKeyGen(true);
+ metaClient = HoodieTestUtils.init(storageConf, basePath,
HoodieTableType.COPY_ON_WRITE, props);
+ config = HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withProps(props)
+ .withParallelism(2, 2)
+ .withBulkInsertParallelism(2)
+ .withFinalizeWriteParallelism(2)
+ .withDeleteParallelism(2)
+ .forTable("external_file_clustering_table")
+ // The write handle creates its marker file in the constructor, so
keep markers direct and
+ // avoid the embedded timeline server.
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
+
.withStorageConfig(HoodieStorageConfig.newBuilder().parquetMaxFileSize(1024 *
1024).build())
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ // Single file sort plan strategy emits exactly one file slice per
clustering group,
+ // which is what the external file strategy requires.
+
.withClusteringPlanStrategyClass(SparkSingleFileSortPlanStrategy.class.getName())
+
.withClusteringExecutionStrategyClass(ExternalFileClusteringTestExecutionStrategy.class.getName())
+ .build())
+ .build();
+ writeClient = getHoodieWriteClient(config);
+ }
+
+ @AfterEach
+ public void tearDown() throws IOException {
+ cleanupResources();
+ }
+
+ /**
+ * Schedules a real clustering plan over two file groups and runs every
group of that plan through
+ * the strategy, asserting that each clustered file is a faithful copy of
its input and that the
+ * write status the handle derives from it is complete.
+ *
+ * <p>NOTE: the groups are executed through {@code
performClusteringForGroup} rather than through
+ * {@code SparkRDDWriteClient#cluster}, because {@code
SingleSparkJobExecutionStrategy#performClustering}
+ * eagerly builds a Spark reader context, which needs the engine specific
+ * {@code org.apache.spark.sql.adapter.*Adapter} that only lives in the
downstream
+ * hudi-spark-datasource modules and hence is not on this module's test
classpath.
+ */
+ @Test
+ public void testClusteringCopiesFileContentsAndPreservesMetadata() throws
Exception {
+ List<HoodieRecord> records = new ArrayList<>();
+ records.addAll(bulkInsertBatch(60));
+ records.addAll(bulkInsertBatch(40));
+ List<HoodieBaseFile> sourceFiles = listBaseFiles();
+ assertEquals(2, sourceFiles.size(), "Each bulk insert should have written
its own file group");
+
+ String clusteringInstant = (String)
writeClient.scheduleClustering(Option.empty()).get();
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieClusteringPlan plan = ClusteringUtils.getClusteringPlan(
+ metaClient,
INSTANT_GENERATOR.getClusteringCommitRequestedInstant(clusteringInstant)).map(Pair::getRight).get();
+ // Both preconditions of the strategy have to be met by the plan it is
paired with.
+ assertTrue(plan.getPreserveHoodieMetadata(), "The plan strategy must
preserve the Hudi metadata fields");
+ assertEquals(2, plan.getInputGroups().size(), "One clustering group per
input file group");
+
+ SparkExternalFileClusteringExecutionStrategy<HoodieAvroPayload> strategy =
+ new ExternalFileClusteringTestExecutionStrategy<>(createTable(),
context, config);
+ List<StoragePath> clusteredPaths = new ArrayList<>();
+ long totalWrites = 0;
+ for (HoodieClusteringGroup group : plan.getInputGroups()) {
+ ClusteringGroupInfo groupInfo = ClusteringGroupInfo.create(group);
+ assertEquals(1, groupInfo.getOperations().size(), "The strategy only
accepts single operation groups");
+ HoodieBaseFile sourceFile = sourceFiles.stream()
+ .filter(baseFile ->
baseFile.getFileId().equals(groupInfo.getOperations().get(0).getFileId()))
+ .findFirst().get();
+
+ List<WriteStatus> writeStatuses =
strategy.performClusteringForGroup(null, groupInfo,
+ plan.getStrategy().getStrategyParams(),
plan.getPreserveHoodieMetadata(), null,
+ new LocalTaskContextSupplier(), clusteringInstant);
+
+ assertEquals(1, writeStatuses.size(), "One clustering operation produces
one write status");
+ HoodieWriteStat stat = writeStatuses.get(0).getStat();
+ assertEquals(PARTITION_PATH, stat.getPartitionPath());
+ assertEquals(sourceFile.getCommitTime(), stat.getPrevCommit(), "Previous
commit is derived from the input file");
+ assertEquals(stat.getFileSizeInBytes(), stat.getTotalWriteBytes());
+ assertTrue(stat.getFileSizeInBytes() > 0);
+ assertTrue(stat.getRuntimeStats().getTotalCreateTime() >= 0);
+ assertEquals(stat.getNumWrites(), stat.getNumInserts());
+ totalWrites += stat.getNumWrites();
+
+ StoragePath clusteredPath = new StoragePath(metaClient.getBasePath(),
stat.getPath());
+ assertTrue(storage.exists(clusteredPath), "Clustered file must land
under the table base path");
+ assertEquals(clusteringInstant,
FSUtils.getCommitTime(clusteredPath.getName()));
+ assertEquals(stat.getFileId(),
FSUtils.getFileId(clusteredPath.getName()));
+ assertEquals(writeStatuses.get(0).getFileId(), stat.getFileId());
+ assertEquals(stat.getNumWrites(), new
ParquetUtils().getRowCount(storage, clusteredPath));
+ // The transformation is a plain file copy, so each clustered file must
equal its input exactly.
+ assertEquals(readParquetAsJson(sourceFile.getPath()),
readParquetAsJson(clusteredPath.toString()),
+ "Clustered file contents must equal the input file contents");
+ clusteredPaths.add(clusteredPath);
+ }
+ assertEquals(records.size(), totalWrites, "Every input record must be
accounted for");
+
+ // The write handle registers each output file with a CREATE marker.
+ Set<String> markerPaths = WriteMarkersFactory.get(MarkerType.DIRECT,
createTable(), clusteringInstant).allMarkerFilePaths();
+ clusteredPaths.forEach(clusteredPath ->
+ assertTrue(markerPaths.stream().anyMatch(marker ->
marker.startsWith(PARTITION_PATH + "/" + clusteredPath.getName())
+ && marker.endsWith(IOType.CREATE.name())), "Expected a CREATE
marker for " + clusteredPath + ", got " + markerPaths));
+
+ List<Row> clusteredRows = sqlContext.read()
+
.parquet(clusteredPaths.stream().map(StoragePath::toString).toArray(String[]::new)).collectAsList();
+
assertEquals(records.stream().map(HoodieRecord::getRecordKey).collect(Collectors.toSet()),
+ clusteredRows.stream().map(row ->
row.<String>getAs(HoodieRecord.RECORD_KEY_METADATA_FIELD)).collect(Collectors.toSet()));
+ // The metadata columns are copied as is, so the rows still carry the
commit times that inserted them.
+
assertEquals(sourceFiles.stream().map(HoodieBaseFile::getCommitTime).collect(Collectors.toSet()),
+ clusteredRows.stream().map(row ->
row.<String>getAs(HoodieRecord.COMMIT_TIME_METADATA_FIELD)).collect(Collectors.toSet()),
+ "_hoodie_commit_time must not be rewritten by clustering");
+ }
+
+ @Test
+ public void testPerformClusteringForGroupRejectsMetadataRewrite() {
+ SparkExternalFileClusteringExecutionStrategy<HoodieAvroPayload> strategy =
+ new ExternalFileClusteringTestExecutionStrategy<>(createTable(),
context, config);
+ ClusteringGroupInfo groupInfo =
clusteringGroup(sliceInfo(newBaseFileName()));
+
+ HoodieClusteringException exception =
assertThrows(HoodieClusteringException.class,
+ () -> strategy.performClusteringForGroup(null, groupInfo,
Collections.emptyMap(), false, null,
+ new LocalTaskContextSupplier(),
WriteClientTestUtils.createNewInstantTime()));
+ assertTrue(exception.getMessage().contains("preserveHoodieMetadata must be
true"), exception.getMessage());
+ }
+
+ @Test
+ public void testPerformClusteringForGroupRejectsMultipleOperations() {
+ SparkExternalFileClusteringExecutionStrategy<HoodieAvroPayload> strategy =
+ new ExternalFileClusteringTestExecutionStrategy<>(createTable(),
context, config);
+ ClusteringGroupInfo groupInfo =
clusteringGroup(sliceInfo(newBaseFileName()), sliceInfo(newBaseFileName()));
+
+ HoodieClusteringException exception =
assertThrows(HoodieClusteringException.class,
+ () -> strategy.performClusteringForGroup(null, groupInfo,
Collections.emptyMap(), true, null,
+ new LocalTaskContextSupplier(),
WriteClientTestUtils.createNewInstantTime()));
+ assertTrue(exception.getMessage().contains("Expect only one clustering
operation during rewrite"), exception.getMessage());
+ }
+
+ @Test
+ public void testPerformClusteringForGroupCleansUpPartialOutputOnFailure()
throws IOException {
+ FailingTransformStrategy strategy = new
FailingTransformStrategy(createTable(), context, config);
+ String dataFileName = newBaseFileName();
+ ClusteringGroupInfo groupInfo = clusteringGroup(sliceInfo(dataFileName));
+
+ HoodieClusteringException exception =
assertThrows(HoodieClusteringException.class,
+ () -> strategy.performClusteringForGroup(null, groupInfo,
Collections.emptyMap(), true, null,
+ new LocalTaskContextSupplier(),
WriteClientTestUtils.createNewInstantTime()));
+ assertTrue(exception.getMessage().contains("Failed to transform file: "),
exception.getMessage());
+ assertTrue(exception.getMessage().contains(dataFileName),
exception.getMessage());
+ assertNotNull(strategy.outputPath, "Transformation must have been handed
the output path");
+ assertFalse(storage.exists(strategy.outputPath), "Partial output file must
be cleaned up");
+ }
+
+ @Test
+ public void
testPerformClusteringForGroupFailsWhenTransformationWritesNothing() {
+ NoOpTransformStrategy strategy = new NoOpTransformStrategy(createTable(),
context, config);
+ ClusteringGroupInfo groupInfo =
clusteringGroup(sliceInfo(newBaseFileName()));
+
+ HoodieClusteringException exception =
assertThrows(HoodieClusteringException.class,
+ () -> strategy.performClusteringForGroup(null, groupInfo,
Collections.emptyMap(), true, null,
+ new LocalTaskContextSupplier(),
WriteClientTestUtils.createNewInstantTime()));
+ assertTrue(exception.getMessage().contains("Output file does not exist"),
exception.getMessage());
+ }
+
+ private HoodieTable createTable() {
+ return HoodieSparkTable.create(config, context,
HoodieTableMetaClient.reload(metaClient));
+ }
+
+ private String newInstantTime() throws InterruptedException {
+ // Hudi instant times have millisecond resolution, so keep back-to-back
instants strictly ordered.
+ Thread.sleep(2);
+ return WriteClientTestUtils.createNewInstantTime();
+ }
+
+ private List<HoodieRecord> bulkInsertBatch(int numRecords) throws
InterruptedException {
+ String instantTime = newInstantTime();
+ List<HoodieRecord> records =
dataGen.generateInsertsForPartition(instantTime, numRecords, PARTITION_PATH);
+ WriteClientTestUtils.startCommitWithTime(writeClient, instantTime);
+ JavaRDD<WriteStatus> writeStatuses =
writeClient.bulkInsert(jsc.parallelize(records, 1), instantTime);
+ List<WriteStatus> statusList = writeStatuses.collect();
+ assertNoWriteErrors(statusList);
+ writeClient.commit(instantTime, jsc.parallelize(statusList, 1));
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ return records;
+ }
+
+ private List<HoodieBaseFile> listBaseFiles() {
+ return HoodieClientTestUtils.getLatestBaseFiles(basePath, storage,
basePath + "/" + PARTITION_PATH + "/*");
+ }
+
+ private List<String> readParquetAsJson(String path) {
+ List<String> rows = new
ArrayList<>(sqlContext.read().parquet(path).toJSON().collectAsList());
+ Collections.sort(rows);
+ return rows;
+ }
+
+ /**
+ * Builds a Hudi conforming base file name. The write handle derives the
previous commit from the
+ * old file name, so it has to be parseable even when the file itself is
never read.
+ */
+ private String newBaseFileName() {
+ return
FSUtils.makeBaseFileName(WriteClientTestUtils.createNewInstantTime(), "1-0-1",
+ FSUtils.createNewFileIdPfx(),
HoodieFileFormat.PARQUET.getFileExtension());
+ }
+
+ private HoodieSliceInfo sliceInfo(String dataFileName) {
+ return HoodieSliceInfo.newBuilder()
+ .setPartitionPath(PARTITION_PATH)
+ .setFileId(FSUtils.getFileId(dataFileName))
+ .setDataFilePath(basePath + "/" + PARTITION_PATH + "/" + dataFileName)
+ .setDeltaFilePaths(Collections.emptyList())
+ .setBootstrapFilePath("")
+ .build();
+ }
+
+ private static ClusteringGroupInfo clusteringGroup(HoodieSliceInfo...
slices) {
+ return ClusteringGroupInfo.create(HoodieClusteringGroup.newBuilder()
+ .setSlices(Arrays.asList(slices))
+ .setNumOutputFileGroups(1)
+ .setMetrics(new HashMap<>())
+ .build());
+ }
+
+ /**
+ * Writes a partial output file and then fails, to exercise the cleanup path
of the strategy.
+ */
+ private static class FailingTransformStrategy extends
SparkExternalFileClusteringExecutionStrategy<HoodieAvroPayload> {
+
+ private StoragePath outputPath;
+
+ FailingTransformStrategy(HoodieTable table, HoodieEngineContext
engineContext, HoodieWriteConfig writeConfig) {
+ super(table, engineContext, writeConfig);
+ }
+
+ @Override
+ protected void transformFile(StoragePath oldFilePath, StoragePath
newFilePath) {
+ this.outputPath = newFilePath;
+ try (OutputStream outputStream =
getHoodieTable().getStorage().create(newFilePath)) {
+ outputStream.write("partial output".getBytes(StandardCharsets.UTF_8));
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to write partial output file", e);
+ }
+ throw new RuntimeException("transformation failed");
+ }
+ }
+
+ /**
+ * Silently does nothing, to exercise the missing output file check of the
write handle.
+ */
+ private static class NoOpTransformStrategy extends
SparkExternalFileClusteringExecutionStrategy<HoodieAvroPayload> {
+
+ NoOpTransformStrategy(HoodieTable table, HoodieEngineContext
engineContext, HoodieWriteConfig writeConfig) {
+ super(table, engineContext, writeConfig);
+ }
+
+ @Override
+ protected void transformFile(StoragePath oldFilePath, StoragePath
newFilePath) {
+ // intentionally writes nothing
+ }
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/bootstrap/TestOrcBootstrapMetadataHandler.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/bootstrap/TestOrcBootstrapMetadataHandler.java
new file mode 100644
index 000000000000..9c8e7dfee0f9
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/bootstrap/TestOrcBootstrapMetadataHandler.java
@@ -0,0 +1,237 @@
+/*
+ * 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.table.action.bootstrap;
+
+import org.apache.hudi.DefaultSparkRecordMerger;
+import org.apache.hudi.avro.model.HoodieFileStatus;
+import org.apache.hudi.client.bootstrap.BootstrapWriteStatus;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.BootstrapFileMapping;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+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.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.OrcUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieBootstrapConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.keygen.KeyGeneratorInterface;
+import org.apache.hudi.keygen.NonpartitionedKeyGenerator;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.testutils.HoodieClientTestBase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.CompressionKind;
+import org.apache.orc.OrcFile;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link OrcBootstrapMetadataHandler}.
+ *
+ * <p>NOTE: the end-to-end ORC bootstrap coverage lives in {@code
TestOrcBootstrap}, which has been
+ * disabled since HUDI-7353. This test drives the handler directly instead, so
that the ORC arm of
+ * {@link MetadataBootstrapHandlerFactory} keeps being exercised.
+ */
+public class TestOrcBootstrapMetadataHandler extends HoodieClientTestBase {
+
+ private static final String SOURCE_ORC_SCHEMA =
"struct<_row_key:string,rider:string,fare:int>";
+ private static final String RECORD_KEY_FIELD = "_row_key";
+ private static final String SRC_PARTITION_PATH = "2020/04/01";
+ // Deliberately different from the source partition path, the way a
BootstrapPartitionPathTranslator
+ // would rewrite it, so that the two never get transposed unnoticed.
+ private static final String PARTITION_PATH = "2020-04-01";
+ private static final int NUM_SOURCE_RECORDS = 5;
+
+ private String bootstrapBasePath;
+ private StoragePath sourceFilePath;
+ private HoodieFileStatus srcFileStatus;
+
+ @BeforeEach
+ public void initBootstrapSource() throws IOException {
+ bootstrapBasePath =
tempDir.resolve("bootstrap_source").toAbsolutePath().toString();
+ sourceFilePath = new StoragePath(bootstrapBasePath + "/" +
SRC_PARTITION_PATH,
+ "src_0" + HoodieFileFormat.ORC.getFileExtension());
+ writeOrcSourceFile();
+
+ // Re-initialize the target table so that its base file format is ORC and
it points at the ORC source.
+ metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE,
bootstrapBasePath,
+ HoodieFileFormat.ORC,
NonpartitionedKeyGenerator.class.getCanonicalName());
+
+ List<Pair<String, List<HoodieFileStatus>>> leafFolders =
BootstrapUtils.getAllLeafFoldersWithFiles(
+ HoodieFileFormat.ORC, metaClient.getStorage(), bootstrapBasePath,
context);
+ assertEquals(1, leafFolders.size());
+ assertEquals(SRC_PARTITION_PATH, leafFolders.get(0).getKey());
+ srcFileStatus = leafFolders.get(0).getValue().get(0);
+ }
+
+ @Test
+ public void testGetSchemaConvertsOrcTypeDescription() throws IOException {
+ HoodieWriteConfig config = bootstrapConfigBuilder().build();
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ OrcBootstrapMetadataHandler handler = getOrcHandler(config, table);
+
+ HoodieSchema schema = handler.getSchema(sourceFilePath);
+
+ assertEquals(HoodieSchemaType.RECORD, schema.getType());
+ assertEquals(Arrays.asList(RECORD_KEY_FIELD, "rider", "fare"),
+
schema.getFields().stream().map(HoodieSchemaField::name).collect(Collectors.toList()));
+ assertEquals(Arrays.asList(HoodieSchemaType.STRING,
HoodieSchemaType.STRING, HoodieSchemaType.INT),
+ schema.getFields().stream().map(f ->
f.schema().getType()).collect(Collectors.toList()));
+ }
+
+ @Test
+ public void testRunMetadataBootstrapWritesSkeletonFile() throws IOException {
+ HoodieWriteConfig config = bootstrapConfigBuilder().build();
+ assertEquals(HoodieRecord.HoodieRecordType.AVRO,
config.getRecordMerger().getRecordType());
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ // Dispatch is purely by the source file's .orc extension.
+ assertInstanceOf(OrcBootstrapMetadataHandler.class,
+ MetadataBootstrapHandlerFactory.getMetadataHandler(config, table,
srcFileStatus));
+
+ // The handle relies on the spark task context, so run the bootstrap
inside a task the way
+ // SparkBootstrapCommitActionExecutor does.
+ KeyGeneratorInterface keyGenerator = newKeyGenerator();
+ List<BootstrapWriteStatus> writeStatuses = context
+ .parallelize(Collections.singletonList(srcFileStatus), 1)
+ .map(fileStatus ->
MetadataBootstrapHandlerFactory.getMetadataHandler(config, table, fileStatus)
+ .runMetadataBootstrap(SRC_PARTITION_PATH, PARTITION_PATH,
keyGenerator))
+ .collectAsList();
+ assertEquals(1, writeStatuses.size());
+ BootstrapWriteStatus writeStatus = writeStatuses.get(0);
+
+ assertFalse(writeStatus.hasErrors());
+ assertEquals(PARTITION_PATH, writeStatus.getPartitionPath());
+
+ HoodieWriteStat stat = writeStatus.getStat();
+ assertEquals(NUM_SOURCE_RECORDS, stat.getNumWrites());
+ assertEquals(NUM_SOURCE_RECORDS, stat.getNumInserts());
+ assertEquals(0, stat.getTotalWriteErrors());
+ assertEquals(writeStatus.getFileId(), stat.getFileId());
+
+ // The skeleton base file is written in the table's base file format, ie
ORC.
+ StoragePath skeletonPath = new StoragePath(basePath, stat.getPath());
+
assertTrue(skeletonPath.getName().endsWith(HoodieFileFormat.ORC.getFileExtension()));
+ assertTrue(stat.getPath().startsWith(PARTITION_PATH + "/"));
+ assertTrue(metaClient.getStorage().exists(skeletonPath),
+ "Skeleton base file " + skeletonPath + " should exist");
+
+ // Only the record keys of the source file are carried over into the
skeleton file.
+ Set<Pair<String, Long>> writtenKeys =
+ new OrcUtils().filterRowKeys(metaClient.getStorage(), skeletonPath,
Collections.emptySet());
+ assertEquals(expectedRecordKeys(),
+
writtenKeys.stream().map(Pair::getLeft).sorted().collect(Collectors.toList()));
+
+ BootstrapFileMapping mapping = writeStatus.getBootstrapSourceFileMapping();
+ assertEquals(bootstrapBasePath, mapping.getBootstrapBasePath());
+ assertEquals(SRC_PARTITION_PATH, mapping.getBootstrapPartitionPath());
+ assertEquals(srcFileStatus, mapping.getBootstrapFileStatus());
+ assertEquals(PARTITION_PATH, mapping.getPartitionPath());
+ assertEquals(writeStatus.getFileId(), mapping.getFileId());
+ }
+
+ @Test
+ public void testExecuteBootstrapRejectsSparkRecordType() {
+ HoodieWriteConfig config = bootstrapConfigBuilder()
+ .withRecordMergeImplClasses(DefaultSparkRecordMerger.class.getName())
+ .build();
+ assertEquals(HoodieRecord.HoodieRecordType.SPARK,
config.getRecordMerger().getRecordType());
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ OrcBootstrapMetadataHandler handler = getOrcHandler(config, table);
+
+ // The spark reader is not wired up for ORC bootstrap, the handler bails
out before touching any handle.
+ assertThrows(UnsupportedOperationException.class,
+ () -> handler.executeBootstrap(null, sourceFilePath,
newKeyGenerator(), PARTITION_PATH, null));
+ }
+
+ private OrcBootstrapMetadataHandler getOrcHandler(HoodieWriteConfig config,
HoodieTable table) {
+ BootstrapMetadataHandler handler =
+ MetadataBootstrapHandlerFactory.getMetadataHandler(config, table,
srcFileStatus);
+ return assertInstanceOf(OrcBootstrapMetadataHandler.class, handler);
+ }
+
+ private HoodieWriteConfig.Builder bootstrapConfigBuilder() {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .forTable("test_orc_bootstrap")
+ .withEmbeddedTimelineServerEnabled(false)
+ .withWriteStatusClass(BootstrapWriteStatus.class)
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
+ .withBootstrapConfig(HoodieBootstrapConfig.newBuilder()
+ .withBootstrapBasePath(bootstrapBasePath).build());
+ }
+
+ private KeyGeneratorInterface newKeyGenerator() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(),
RECORD_KEY_FIELD);
+ return new NonpartitionedKeyGenerator(props);
+ }
+
+ private static List<String> expectedRecordKeys() {
+ return IntStream.range(0, NUM_SOURCE_RECORDS).mapToObj(i -> "key" +
i).sorted().collect(Collectors.toList());
+ }
+
+ private void writeOrcSourceFile() throws IOException {
+ TypeDescription orcSchema = TypeDescription.fromString(SOURCE_ORC_SCHEMA);
+ OrcFile.WriterOptions options =
OrcFile.writerOptions(storageConf.unwrapAs(Configuration.class))
+ .setSchema(orcSchema).compress(CompressionKind.ZLIB);
+ try (Writer writer = OrcFile.createWriter(new
Path(sourceFilePath.toUri()), options)) {
+ VectorizedRowBatch batch = orcSchema.createRowBatch();
+ BytesColumnVector rowKeys = (BytesColumnVector) batch.cols[0];
+ BytesColumnVector riders = (BytesColumnVector) batch.cols[1];
+ LongColumnVector fares = (LongColumnVector) batch.cols[2];
+ for (int r = 0; r < NUM_SOURCE_RECORDS; r++) {
+ int row = batch.size++;
+ rowKeys.setVal(row, getUTF8Bytes("key" + r));
+ riders.setVal(row, getUTF8Bytes("rider" + r));
+ fares.vector[row] = r;
+ }
+ writer.addRowBatch(batch);
+ }
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
index 1288322b6682..c23c37b16a47 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
@@ -21,9 +21,12 @@ package org.apache.hudi.table.action.commit;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
@@ -40,10 +43,14 @@ import org.apache.hudi.config.HoodieLayoutConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.io.storage.HoodieIOFactory;
import org.apache.hudi.data.HoodieJavaRDD;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.hadoop.HoodieParquetInputFormat;
import org.apache.hudi.hadoop.utils.HoodieHiveUtils;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.io.HoodieCreateHandle;
+import org.apache.hudi.io.HoodieWriteMergeHandle;
+import org.apache.hudi.keygen.KeyGeneratorInterface;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieSparkCopyOnWriteTable;
@@ -91,6 +98,7 @@ import static
org.apache.hudi.common.testutils.HoodieTestTable.makeNewCommitTime
import static
org.apache.hudi.common.testutils.HoodieTestUtils.createSimpleRecord;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -533,6 +541,123 @@ public class TestCopyOnWriteActionExecutor extends
HoodieClientTestBase implemen
assertTrue(partitionMetadata.readPartitionCreatedCommitTime().get().equals(instantTime));
}
+ @Test
+ public void testCompactionIsNotSupportedOnCopyOnWrite() {
+ HoodieWriteConfig config = makeHoodieClientConfig();
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieSparkCopyOnWriteTable table =
+ (HoodieSparkCopyOnWriteTable) HoodieSparkTable.create(config, context,
metaClient);
+
+ assertThrows(HoodieNotSupportedException.class,
+ () -> table.scheduleCompaction(context, makeNewCommitTime(),
Option.empty()));
+ assertThrows(HoodieNotSupportedException.class, () ->
table.compact(context, makeNewCommitTime()));
+ }
+
+ /**
+ * {@link org.apache.hudi.table.HoodieCompactionHandler#handleInsert} is the
map-based entry point used by the
+ * compactor, so it is not reached by a regular insert through the write
client.
+ */
+ @Test
+ public void testHandleInsertThroughTheCompactionHandler() throws Exception {
+ HoodieWriteConfig config = makeHoodieClientConfig();
+ SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+ String instantTime = writeClient.startCommit();
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieSparkCopyOnWriteTable table =
+ (HoodieSparkCopyOnWriteTable) HoodieSparkTable.create(config, context,
metaClient);
+
+ // The write handle needs a live Spark TaskContext to build its write
token, so drive it from inside a task.
+ // The records are built there too: they carry their schema by reference
and do not survive a round trip.
+ int numRecords = 3;
+ List<WriteStatus> statuses = jsc.parallelize(Arrays.asList(1), 1)
+ .map(x -> {
+ Map<String, HoodieRecord<?>> recordMap = new HashMap<>();
+ for (int i = 0; i < numRecords; i++) {
+ HoodieRecord record = createSimpleRecord("key-" + i,
"2016-01-31T03:16:41.415Z", i);
+ recordMap.put(record.getRecordKey(), record);
+ }
+ return table.handleInsert(instantTime, "2016/01/31",
FSUtils.createNewFileIdPfx(), recordMap);
+ })
+ .flatMap(Transformations::flattenAsIterator).collect();
+
+ assertEquals(1, statuses.size());
+ assertFalse(statuses.get(0).hasErrors());
+ assertEquals(numRecords, statuses.get(0).getStat().getNumWrites());
+ }
+
+ /**
+ * {@link org.apache.hudi.table.HoodieCompactionHandler#handleUpdate} builds
the merge handle through
+ * {@link
org.apache.hudi.table.HoodieSparkCopyOnWriteTable#getUpdateHandle}, which
rejects key generators that do
+ * not extend BaseKeyGenerator once the meta fields are switched off.
+ */
+ @Test
+ public void
testGetUpdateHandleRejectsNonBaseKeyGeneratorWhenMetaFieldsAreDisabled() throws
Exception {
+ Properties props = new Properties();
+ props.setProperty(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
HoodieWriteMergeHandle.class.getName());
+ props.setProperty(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ props.setProperty(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key(),
NonBaseKeyGenerator.class.getName());
+ HoodieWriteConfig config =
makeHoodieClientConfigBuilder().withProps(props).build();
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieSparkCopyOnWriteTable table =
+ (HoodieSparkCopyOnWriteTable) HoodieSparkTable.create(config, context,
metaClient);
+
+ HoodieRecord updated = createSimpleRecord("key-0",
"2016-01-31T03:16:41.415Z", 99);
+ Map<String, HoodieRecord<?>> keyToNewRecords =
+ Collections.singletonMap(updated.getRecordKey(), updated);
+
+ assertThrows(HoodieException.class, () -> table.handleUpdate(
+ makeNewCommitTime(), "2016/01/31", FSUtils.createNewFileIdPfx(),
keyToNewRecords, null));
+ }
+
+ @Test
+ public void testRollbackBootstrapRestoresTableToInitialState() throws
Exception {
+ // The metadata table is rebuilt by a bootstrap re-attempt, and restoring
it back past its own first base file
+ // is not supported, so keep it out of this test.
+ Properties props = new Properties();
+ props.setProperty(HoodieMetadataConfig.ENABLE.key(), "false");
+ HoodieWriteConfig config =
makeHoodieClientConfigBuilder().withProps(props).build();
+ SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+
+ String instantTime = writeClient.startCommit();
+ List<HoodieRecord> records = Collections.singletonList(
+ createSimpleRecord("key-0", "2016-01-31T03:16:41.415Z", 10));
+ JavaRDD<WriteStatus> writeStatuses =
writeClient.insert(jsc.parallelize(records, 1), instantTime);
+ writeClient.commit(instantTime, writeStatuses, Option.empty(),
COMMIT_ACTION,
+ Collections.emptyMap(), Option.empty());
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
assertFalse(metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().empty());
+
+ HoodieSparkCopyOnWriteTable table =
+ (HoodieSparkCopyOnWriteTable) HoodieSparkTable.create(config, context,
metaClient);
+ table.rollbackBootstrap(context, makeNewCommitTime());
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
assertTrue(metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().empty());
+ }
+
+ /**
+ * A key generator that deliberately does not extend {@link
org.apache.hudi.keygen.BaseKeyGenerator}, which is the
+ * only shape supported when the meta fields are turned off.
+ */
+ public static class NonBaseKeyGenerator implements KeyGeneratorInterface {
+
+ public NonBaseKeyGenerator(TypedProperties props) {
+ // The config only has to be loadable; it is rejected before it is ever
used.
+ }
+
+ @Override
+ public HoodieKey getKey(GenericRecord record) {
+ return new HoodieKey(record.get("_row_key").toString(), "2016/01/31");
+ }
+
+ @Override
+ public List<String> getRecordKeyFieldNames() {
+ return Collections.singletonList("_row_key");
+ }
+ }
+
// methods below were copied from [[TestBulkInsertInternalPartitioner]]
public static JavaRDD<HoodieRecord>
generateTestRecordsForBulkInsert(JavaSparkContext jsc) {
HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator();
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBootstrapRead.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBootstrapRead.java
index 1e36f491b3f6..6be6ed4676e2 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBootstrapRead.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBootstrapRead.java
@@ -19,11 +19,14 @@
package org.apache.hudi.functional;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.HoodieWriteMergeHandle;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -96,4 +99,44 @@ public class TestBootstrapRead extends TestBootstrapReadBase
{
compareTables();
verifyMetaColOnlyRead(2);
}
+
+ /**
+ * Same metadata-bootstrap COW scenario as {@link #testBootstrapFunctional},
but pinned to the legacy
+ * {@link HoodieWriteMergeHandle}. That handle is the only one routed through
+ * {@code HoodieMergeHelper#runMerge}, so this is what exercises the
bootstrap arm of the merge helper
+ * (it stitches the skeleton file and the bootstrap base file together via a
bootstrap file reader).
+ * The default merge handle is {@code FileGroupReaderBasedMergeHandle},
which overrides {@code doMerge}
+ * and never reaches the merge helper.
+ */
+ @Test
+ public void testBootstrapUpsertWithLegacyMergeHandle() {
+ this.bootstrapType = "metadata";
+ this.dashPartitions = true;
+ this.tableType = COPY_ON_WRITE;
+ this.nPartitions = 0;
+ setupDirs();
+
+ // do bootstrap
+ Map<String, String> options = withLegacyMergeHandle(setBootstrapOptions());
+ Dataset<Row> bootstrapDf = sparkSession.emptyDataFrame();
+ bootstrapDf.write().format("hudi")
+ .options(options)
+ .mode(SaveMode.Overwrite)
+ .save(bootstrapTargetPath);
+ compareTables();
+ verifyMetaColOnlyRead(0);
+
+ // upsert into the bootstrapped file groups, so their base files get merged
+ options = withLegacyMergeHandle(basicOptions());
+ doUpdate(options, "001");
+ compareTables();
+ verifyMetaColOnlyRead(1);
+ }
+
+ private static Map<String, String> withLegacyMergeHandle(Map<String, String>
options) {
+ options.put(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
HoodieWriteMergeHandle.class.getName());
+ // do not silently fall back to the default merge handle if the legacy one
cannot be instantiated
+ options.put(HoodieWriteConfig.MERGE_HANDLE_PERFORM_FALLBACK.key(),
"false");
+ return options;
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkBinaryCopyClusteringAndValidationMeta.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkBinaryCopyClusteringAndValidationMeta.java
index f3a9be923ff7..9595edbf47d1 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkBinaryCopyClusteringAndValidationMeta.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkBinaryCopyClusteringAndValidationMeta.java
@@ -22,6 +22,7 @@ import org.apache.hudi.avro.HoodieAvroWriteSupport;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
import
org.apache.hudi.client.clustering.run.strategy.SparkBinaryCopyClusteringExecutionStrategy;
+import
org.apache.hudi.client.clustering.run.strategy.SparkStreamCopyClusteringExecutionStrategy;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.avro.HoodieBloomFilterWriteSupport;
import org.apache.hudi.common.bloom.BloomFilter;
@@ -32,6 +33,7 @@ import org.apache.hudi.common.config.HoodieParquetConfig;
import org.apache.hudi.common.engine.LocalTaskContextSupplier;
import org.apache.hudi.common.model.ClusteringGroupInfo;
import org.apache.hudi.common.model.ClusteringOperation;
+import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
import org.apache.hudi.common.model.HoodieTableType;
@@ -87,6 +89,7 @@ import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.AVRO_SCHE
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.HOODIE_SCHEMA;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_NESTED_EXAMPLE_SCHEMA;
+import static
org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -146,6 +149,18 @@ public class
TestSparkBinaryCopyClusteringAndValidationMeta extends HoodieClient
client.commit(newCommitTime2, statuses2);
List<WriteStatus> statusList2 = statuses2.collect();
+ Set<String> commitTimeSet = new HashSet<>();
+ commitTimeSet.add(newCommitTime1);
+ commitTimeSet.add(newCommitTime2);
+ checkClusteredFilesAgainstReplaceCommit(partitionPath, allRecord,
commitTimeSet);
+ }
+
+ /**
+ * Finds all parquet files created as part of clustering, verifies they
match what is found in the replace commit
+ * metadata and that their footers describe the given records. Returns the
clustered files, relative to the base path.
+ */
+ private List<String> checkClusteredFilesAgainstReplaceCommit(String
partitionPath, List<HoodieRecord> allRecord,
+ Set<String>
commitTimeSet) throws IOException {
metaClient = HoodieTableMetaClient.reload(metaClient);
HoodieInstant replaceCommitInstant = metaClient.getActiveTimeline()
.getCompletedReplaceTimeline().firstInstant().get();
@@ -156,25 +171,18 @@ public class
TestSparkBinaryCopyClusteringAndValidationMeta extends HoodieClient
replaceCommitMetadata.getPartitionToWriteStats()
.forEach((k, v) -> v.forEach(entry ->
filesFromReplaceCommit.add(entry.getPath())));
- // find all parquet files created as part of clustering. Verify it matces
w/ whats found in replace commit metadata.
FileStatus[] fileStatuses = fs.listStatus(new Path(basePath + "/" +
partitionPath));
- List<String> replacedFiles = new ArrayList<>();
List<String> clusteredFiles = new ArrayList<>();
String clusteredFileName = "";
for (FileStatus status : fileStatuses) {
if
(status.getPath().getName().contains(replaceCommitInstant.requestedTime())) {
clusteredFiles.add(partitionPath + "/" + status.getPath().getName());
clusteredFileName = status.getPath().getName();
- } else if (!status.getPath().getName().startsWith(".")) {
- replacedFiles.add(partitionPath + "/" + status.getPath().getName());
}
}
assertEquals(clusteredFiles, filesFromReplaceCommit);
- // clusteredFiles check
- Set<String> commitTimeSet = new HashSet();
- commitTimeSet.add(newCommitTime1);
- commitTimeSet.add(newCommitTime2);
checkFileFooter(clusteredFiles, allRecord, commitTimeSet,
clusteredFileName);
+ return clusteredFiles;
}
private void checkFileFooter(List<String> clusteredFiles, List<HoodieRecord>
allRecord,
@@ -236,11 +244,11 @@ public class
TestSparkBinaryCopyClusteringAndValidationMeta extends HoodieClient
MessageType standardSchema = new
AvroSchemaConverter(conf).convert(AVRO_SCHEMA);
BloomFilter simpleBloomFilter = BloomFilterFactory.createBloomFilter(1000,
0.0001, 10000, SIMPLE.name());
- BloomFilter dynmicBloomFilter = BloomFilterFactory.createBloomFilter(1000,
0.0001, 10000, DYNAMIC_V0.name());
+ BloomFilter dynamicBloomFilter =
BloomFilterFactory.createBloomFilter(1000, 0.0001, 10000, DYNAMIC_V0.name());
String file1 = makeTestFile("file-1.parquet", HOODIE_SCHEMA, legacySchema,
simpleBloomFilter);
- String file2 = makeTestFile("file-2.parquet", HOODIE_SCHEMA, legacySchema,
dynmicBloomFilter);
- String file3 = makeTestFile("file-3.parquet", HOODIE_SCHEMA,
standardSchema, dynmicBloomFilter);
- String file4 = makeTestFile("file-4.parquet", HOODIE_SCHEMA,
standardSchema, dynmicBloomFilter);
+ String file2 = makeTestFile("file-2.parquet", HOODIE_SCHEMA, legacySchema,
dynamicBloomFilter);
+ String file3 = makeTestFile("file-3.parquet", HOODIE_SCHEMA,
standardSchema, dynamicBloomFilter);
+ String file4 = makeTestFile("file-4.parquet", HOODIE_SCHEMA,
standardSchema, dynamicBloomFilter);
// input file contains multiple bloom filter code type, should return false
List<ClusteringGroupInfo> groups = makeClusteringGroup(file1, file2);
@@ -260,6 +268,115 @@ public class
TestSparkBinaryCopyClusteringAndValidationMeta extends HoodieClient
Assertions.assertFalse(strategy.supportBinaryStreamCopy(groups, new
HashMap<>()));
}
+ @Test
+ public void testSupportStreamCopy() throws IOException {
+ HoodieWriteConfig writeConfig = new HoodieWriteConfig.Builder()
+ .withPath(basePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withEmbeddedTimelineServerEnabled(false)
+ .build();
+ HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
+ HoodieTable table = HoodieSparkTable.create(writeConfig, engineContext,
metaClient);
+ SparkStreamCopyClusteringExecutionStrategy strategy =
+ new SparkStreamCopyClusteringExecutionStrategy(table, engineContext,
writeConfig);
+
+ MessageType legacySchema = new AvroSchemaConverter().convert(AVRO_SCHEMA);
+ Configuration conf = new Configuration();
+ conf.set("parquet.avro.write-old-list-structure", "false");
+ MessageType standardSchema = new
AvroSchemaConverter(conf).convert(AVRO_SCHEMA);
+
+ BloomFilter simpleBloomFilter = BloomFilterFactory.createBloomFilter(1000,
0.0001, 10000, SIMPLE.name());
+ BloomFilter dynamicBloomFilter =
BloomFilterFactory.createBloomFilter(1000, 0.0001, 10000, DYNAMIC_V0.name());
+ // differs from the others in both bloom filter type code and list
structure
+ String file1 = makeTestFile("stream-copy-1.parquet", HOODIE_SCHEMA,
legacySchema, simpleBloomFilter);
+ String file2 = makeTestFile("stream-copy-2.parquet", HOODIE_SCHEMA,
standardSchema, dynamicBloomFilter);
+ String file3 = makeTestFile("stream-copy-3.parquet", HOODIE_SCHEMA,
standardSchema, dynamicBloomFilter);
+ List<ClusteringGroupInfo> incompatibleGroups = makeClusteringGroup(file1,
file2);
+ List<ClusteringGroupInfo> compatibleGroups = makeClusteringGroup(file2,
file3);
+
+ // schema evolution is disabled by default, so the files are never scanned
and even a mismatch is accepted
+ Assertions.assertTrue(strategy.supportBinaryStreamCopy(incompatibleGroups,
new HashMap<>()));
+
+ // sorting is not supported by binary copy
+ Map<String, String> sortParams = new HashMap<>();
+ sortParams.put(PLAN_STRATEGY_SORT_COLUMNS.key(), "rider");
+ Assertions.assertFalse(strategy.supportBinaryStreamCopy(compatibleGroups,
sortParams));
+
+ // with schema evolution enabled the files are scanned for real
+ HoodieWriteConfig evolutionConfig = new HoodieWriteConfig.Builder()
+ .withPath(basePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withEmbeddedTimelineServerEnabled(false)
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withFileStitchingBinaryCopySchemaEvolutionEnabled(true)
+ .build())
+ .build();
+ SparkStreamCopyClusteringExecutionStrategy evolutionStrategy =
+ new SparkStreamCopyClusteringExecutionStrategy(table, engineContext,
evolutionConfig);
+
Assertions.assertTrue(evolutionStrategy.supportBinaryStreamCopy(compatibleGroups,
new HashMap<>()));
+
Assertions.assertFalse(evolutionStrategy.supportBinaryStreamCopy(incompatibleGroups,
new HashMap<>()));
+
+ // only parquet base files are supported
+ metaClient = HoodieTestUtils.init(basePath, HoodieFileFormat.ORC);
+ HoodieTable orcTable = HoodieSparkTable.create(writeConfig, engineContext,
metaClient);
+ Assertions.assertFalse(new
SparkStreamCopyClusteringExecutionStrategy(orcTable, engineContext, writeConfig)
+ .supportBinaryStreamCopy(compatibleGroups, new HashMap<>()));
+
+ // only COW tables are supported
+ metaClient = HoodieTestUtils.init(basePath, HoodieTableType.MERGE_ON_READ);
+ HoodieTable morTable = HoodieSparkTable.create(writeConfig, engineContext,
metaClient);
+ Assertions.assertFalse(new
SparkStreamCopyClusteringExecutionStrategy(morTable, engineContext, writeConfig)
+ .supportBinaryStreamCopy(compatibleGroups, new HashMap<>()));
+ }
+
+ @Test
+ public void testStreamCopyClusteringEndToEnd() throws IOException {
+ String partitionPath = "2015/03/16";
+ Properties properties = new Properties();
+ properties.setProperty("hoodie.parquet.small.file.limit", "-1");
+ HoodieWriteConfig.Builder cfgBuilder = new HoodieWriteConfig.Builder()
+ .withPath(basePath)
+ .withSchema(TRIP_NESTED_EXAMPLE_SCHEMA)
+ .withEmbeddedTimelineServerEnabled(false)
+ .withClusteringConfig(
+ HoodieClusteringConfig
+ .newBuilder()
+ .withInlineClustering(true)
+ .withAsyncClustering(false)
+ .withInlineClusteringNumCommits(2)
+ // the execution strategy is picked from the write config
rather than from the plan, so set both
+
.withClusteringPlanStrategyClass(HoodieClusteringConfig.SPARK_STREAM_COPY_CLUSTERING_PLAN_STRATEGY)
+
.withClusteringExecutionStrategyClass(HoodieClusteringConfig.SPARK_STREAM_COPY_CLUSTERING_EXECUTION_STRATEGY)
+ .withFileStitchingBinaryCopySchemaEvolutionEnabled(false)
+ .build()).withProps(properties);
+ SparkRDDWriteClient client = getHoodieWriteClient(cfgBuilder.build());
+ HoodieTestDataGenerator dataGen = new
HoodieTestDataGenerator(HoodieTestDataGenerator.TRIP_NESTED_EXAMPLE_SCHEMA,
+ 0xDEED, new String[] {partitionPath}, new HashMap<>());
+
+ List<HoodieRecord> allRecord = new ArrayList<>();
+ Set<String> commitTimeSet = new HashSet<>();
+ // the second commit triggers the inline clustering of both base files
+ for (int i = 0; i < 2; i++) {
+ String newCommitTime = client.startCommit("commit");
+ commitTimeSet.add(newCommitTime);
+ List<HoodieRecord> hoodieRecords =
dataGen.generateInsertsNestedExample(newCommitTime, 30);
+ allRecord.addAll(hoodieRecords);
+ JavaRDD<WriteStatus> statuses =
client.insert(jsc.parallelize(hoodieRecords, 1), newCommitTime);
+ client.commit(newCommitTime, statuses);
+ }
+
+ List<String> clusteredFiles =
checkClusteredFilesAgainstReplaceCommit(partitionPath, allRecord,
commitTimeSet);
+
+ // every input record has to survive the binary stream copy, exactly once
+ List<String> clusteredKeys = sqlContext.read().format("parquet")
+ .load(basePath + "/" + clusteredFiles.get(0))
+ .select("_hoodie_record_key").collectAsList()
+ .stream().map(row -> row.getString(0)).collect(Collectors.toList());
+ assertEquals(allRecord.size(), clusteredKeys.size());
+
assertEquals(allRecord.stream().map(HoodieRecord::getRecordKey).collect(Collectors.toSet()),
+ new HashSet<>(clusteredKeys));
+ }
+
private String makeTestFile(String fileName, HoodieSchema schema,
MessageType messageType, BloomFilter filter) throws IOException {
HoodieAvroWriteSupport writeSupport = new
HoodieAvroWriteSupport(messageType, schema, Option.of(filter), new
Properties());
StoragePath filePath = new
StoragePath(tempDir.resolve(fileName).toAbsolutePath().toString());
