This is an automated email from the ASF dual-hosted git repository.
danny0405 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 260ee6be950c test(common): improve ORC and filesystem utility coverage
(#19542)
260ee6be950c is described below
commit 260ee6be950ccea7107ea82ed2983a5d060972f0
Author: Shuo Cheng <[email protected]>
AuthorDate: Tue Aug 11 10:44:08 2026 +0800
test(common): improve ORC and filesystem utility coverage (#19542)
* test(common): improve ORC and filesystem utility coverage
---
.../apache/hudi/common/util/TestAvroOrcUtils.java | 246 +++++++++++
.../org/apache/hudi/common/util/TestOrcUtils.java | 150 +++++++
.../org/apache/hudi/hadoop/fs/TestCachingPath.java | 105 +++++
.../apache/hudi/hadoop/fs/TestHadoopFSUtils.java | 130 ++++++
.../fs/TestHoodieRetryWrapperFileSystem.java | 158 +++++++
.../fs/TestHoodieWrapperFileSystemOperations.java | 476 +++++++++++++++++++++
.../hadoop/TestHoodieVariantReconstruction.java | 90 ++++
.../io/TestHoodieParquetBinaryCopyBasePaths.java | 76 ++++
8 files changed, 1431 insertions(+)
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java
index f2d4db28809d..8d5d79f82d53 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java
@@ -19,21 +19,44 @@
package org.apache.hudi.common.util;
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.HoodieCommonTestHarness;
+import org.apache.avro.Conversions;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.util.Utf8;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.UnionColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.orc.TypeDescription;
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;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.sql.Date;
import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
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_SCHEMA;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Tests {@link AvroOrcUtils}.
@@ -113,4 +136,227 @@ public class TestAvroOrcUtils extends
HoodieCommonTestHarness {
assertEquals(TypeDescription.Category.LONG,
orcLocalTimestampMicros.getCategory(),
"LocalTimestampMicros should convert to ORC Long (preserving old
behavior)");
}
+
+ @Test
+ public void testPrimitiveValuesRoundTripThroughColumnVectors() {
+ assertTrue((Boolean)
roundTrip(HoodieSchema.create(HoodieSchemaType.BOOLEAN), true));
+ assertEquals(12, roundTrip(HoodieSchema.create(HoodieSchemaType.INT), 12));
+ assertEquals(34L, roundTrip(HoodieSchema.create(HoodieSchemaType.LONG),
34L));
+ assertEquals(1.25f, roundTrip(HoodieSchema.create(HoodieSchemaType.FLOAT),
1.25f));
+ assertEquals(2.5d, roundTrip(HoodieSchema.create(HoodieSchemaType.DOUBLE),
2.5d));
+ assertEquals("hoodie",
roundTrip(HoodieSchema.create(HoodieSchemaType.STRING), new
Utf8("hoodie")).toString());
+ assertEquals(42, roundTrip(HoodieSchema.createDate(), 42));
+ assertEquals(1_700_000_000_123L,
roundTrip(HoodieSchema.createTimestampMillis(), 1_700_000_000_123L));
+ assertEquals(1_700_000_000_123_456L,
+ roundTrip(HoodieSchema.createTimestampMicros(),
1_700_000_000_123_456L));
+
+ ByteBuffer binary = (ByteBuffer) roundTrip(
+ HoodieSchema.create(HoodieSchemaType.BYTES), ByteBuffer.wrap(new
byte[] {1, 2, 3}));
+ assertArrayEquals(new byte[] {1, 2, 3}, toByteArray(binary));
+ }
+
+ @Test
+ public void testComplexValuesRoundTripThroughColumnVectors() {
+ HoodieSchema arraySchema =
HoodieSchema.createArray(HoodieSchema.create(HoodieSchemaType.INT));
+ assertEquals(Arrays.asList(1, 2, 3), roundTrip(arraySchema,
Arrays.asList(1, 2, 3)));
+
+ HoodieSchema mapSchema =
HoodieSchema.createMap(HoodieSchema.create(HoodieSchemaType.LONG));
+ Map<String, Long> values = new LinkedHashMap<>();
+ values.put("first", 10L);
+ values.put("second", 20L);
+ assertEquals(values, roundTrip(mapSchema, values));
+
+ HoodieSchema recordSchema = HoodieSchema.createRecord("nested", null,
null, Arrays.asList(
+ HoodieSchemaField.of("name",
HoodieSchema.create(HoodieSchemaType.STRING), null, null),
+ HoodieSchemaField.of("count",
HoodieSchema.create(HoodieSchemaType.INT), null, null)));
+ GenericRecord record = new GenericData.Record(recordSchema.toAvroSchema());
+ record.put("name", "record-name");
+ record.put("count", 7);
+ GenericRecord converted = (GenericRecord) roundTrip(recordSchema, record);
+ assertEquals("record-name", converted.get("name").toString());
+ assertEquals(7, converted.get("count"));
+
+ HoodieSchema unionSchema = HoodieSchema.createUnion(Arrays.asList(
+ HoodieSchema.create(HoodieSchemaType.INT),
HoodieSchema.create(HoodieSchemaType.STRING)));
+ assertEquals(9, roundTrip(unionSchema, 9));
+ assertEquals("union-value", roundTrip(unionSchema,
"union-value").toString());
+ }
+
+ @Test
+ public void testDecimalEnumAndFixedValuesRoundTripThroughColumnVectors() {
+ HoodieSchema decimalSchema = HoodieSchema.createDecimal(10, 2);
+ ByteBuffer decimalBytes = (ByteBuffer) roundTrip(decimalSchema, new
BigDecimal("1234.50"));
+ assertEquals(new BigInteger("123450"), new
BigInteger(toByteArray(decimalBytes)));
+
+ HoodieSchema fixedDecimalSchema =
HoodieSchema.parse("{\"type\":\"fixed\",\"name\":\"amount\","
+ +
"\"size\":8,\"logicalType\":\"decimal\",\"precision\":12,\"scale\":2}");
+ BigDecimal fixedValue = new BigDecimal("9876.50");
+ GenericData.Fixed fixed = (GenericData.Fixed) new
Conversions.DecimalConversion().toFixed(
+ fixedValue, fixedDecimalSchema.toAvroSchema(),
fixedDecimalSchema.toAvroSchema().getLogicalType());
+ GenericData.Fixed convertedFixed = (GenericData.Fixed)
roundTrip(fixedDecimalSchema, fixed);
+ assertEquals(fixedValue, new Conversions.DecimalConversion().fromFixed(
+ convertedFixed, fixedDecimalSchema.toAvroSchema(),
fixedDecimalSchema.toAvroSchema().getLogicalType()));
+
+ HoodieSchema enumSchema = HoodieSchema.parse(
+
"{\"type\":\"enum\",\"name\":\"status\",\"symbols\":[\"OPEN\",\"CLOSED\"]}");
+ GenericData.EnumSymbol open = new
GenericData.EnumSymbol(enumSchema.toAvroSchema(), "OPEN");
+ assertEquals(open, roundTrip(enumSchema, open));
+ }
+
+ @Test
+ public void testNullResizeAndRepeatingVectors() {
+ HoodieSchema schema = HoodieSchema.create(HoodieSchemaType.LONG);
+ TypeDescription type = AvroOrcUtils.createOrcSchema(schema);
+ VectorizedRowBatch batch =
TypeDescription.createStruct().addField("value", type).createRowBatch();
+ ColumnVector vector = batch.cols[0];
+
+ int expandedPosition = vector.isNull.length;
+ AvroOrcUtils.addToVector(type, vector, schema, null, expandedPosition);
+ assertFalse(vector.noNulls);
+ assertTrue(vector.isNull[expandedPosition]);
+ assertNull(AvroOrcUtils.readFromVector(type, vector, schema,
expandedPosition));
+
+ ((LongColumnVector) vector).vector[0] = 99;
+ vector.isNull[0] = false;
+ vector.isRepeating = true;
+ assertEquals(99L, AvroOrcUtils.readFromVector(type, vector, schema, 17));
+ }
+
+ @Test
+ public void testCreateSchemaCoversEveryOrcCategory() {
+ TypeDescription orcSchema = TypeDescription.fromString("struct<"
+ +
"boolean_field:boolean,byte_field:tinyint,short_field:smallint,int_field:int,long_field:bigint,"
+ +
"float_field:float,double_field:double,string_field:string,char_field:char(5),varchar_field:varchar(8),"
+ +
"date_field:date,timestamp_field:timestamp,binary_field:binary,decimal_field:decimal(12,3),"
+ +
"list_field:array<int>,map_field:map<string,bigint>,union_field:uniontype<int,string>>");
+
+ HoodieSchema schema = AvroOrcUtils.createSchema(orcSchema);
+ assertEquals(HoodieSchemaType.RECORD, schema.getType());
+ assertEquals(17, schema.getFields().size());
+ assertEquals(HoodieSchemaType.BOOLEAN,
schema.getField("boolean_field").get().schema().getType());
+ assertEquals(HoodieSchemaType.BYTES,
schema.getField("binary_field").get().schema().getType());
+ assertEquals(HoodieSchemaType.ARRAY,
schema.getField("list_field").get().schema().getType());
+ assertEquals(HoodieSchemaType.MAP,
schema.getField("map_field").get().schema().getType());
+ assertEquals(HoodieSchemaType.UNION,
schema.getField("union_field").get().schema().getType());
+ }
+
+ @Test
+ public void testCreateSchemaWithDefaultsAndNestedNamespaces() {
+ TypeDescription orcSchema = TypeDescription.fromString(
+
"struct<id:bigint,nested:struct<name:string,active:boolean>,amount:decimal(8,2),tags:array<string>>");
+
+ HoodieSchema nullable = AvroOrcUtils.createSchemaWithDefaultValue(
+ orcSchema, "root_record", "org.apache.hudi.test", true);
+ assertEquals("root_record", nullable.getName());
+ assertTrue(nullable.getFields().stream().allMatch(field ->
field.schema().getType() == HoodieSchemaType.UNION));
+
+ HoodieSchema required = AvroOrcUtils.createSchemaWithDefaultValue(
+ orcSchema, "root_record", "org.apache.hudi.test", false);
+ assertTrue(required.getFields().stream().noneMatch(field ->
field.schema().getType() == HoodieSchemaType.UNION));
+ assertEquals("nested",
required.getField("nested").get().schema().getName());
+ }
+
+ @Test
+ public void testCachingOfByteReferencesDoesNotCopyInput() {
+ HoodieSchema schema = HoodieSchema.create(HoodieSchemaType.BYTES);
+ TypeDescription type = AvroOrcUtils.createOrcSchema(schema);
+ VectorizedRowBatch batch =
TypeDescription.createStruct().addField("value", type).createRowBatch();
+ byte[] input = new byte[] {4, 5, 6};
+
+ AvroOrcUtils.addToVector(type, batch.cols[0], schema, input, 0);
+
+ BytesColumnVector vector = (BytesColumnVector) batch.cols[0];
+ assertSame(input, vector.vector[0]);
+ }
+
+ @Test
+ public void testAdditionalOrcVectorRepresentationsAndFailures() {
+ assertEquals((byte) 7, roundTrip(TypeDescription.createByte(),
HoodieSchema.create(HoodieSchemaType.INT), (byte) 7));
+ assertEquals((short) 11, roundTrip(TypeDescription.createShort(),
HoodieSchema.create(HoodieSchemaType.INT), (short) 11));
+ assertEquals("char",
roundTrip(TypeDescription.createChar().withMaxLength(8),
+ HoodieSchema.create(HoodieSchemaType.STRING), "char"));
+ assertEquals("varchar",
roundTrip(TypeDescription.createVarchar().withMaxLength(8),
+ HoodieSchema.create(HoodieSchemaType.STRING), "varchar"));
+
+ HoodieSchema dateSchema = HoodieSchema.create(HoodieSchemaType.INT);
+ assertEquals(1, roundTrip(TypeDescription.createDate(), dateSchema,
Date.valueOf("1970-01-02")));
+ java.util.Date utilDate = new
java.util.Date(Date.valueOf("1970-01-03").getTime());
+ assertEquals(2, roundTrip(TypeDescription.createDate(), dateSchema,
utilDate));
+
+ HoodieSchema stringSchema = HoodieSchema.create(HoodieSchemaType.STRING);
+ assertThrows(IllegalStateException.class,
+ () -> addValue(TypeDescription.createString(), stringSchema, new
Object()));
+ assertThrows(IllegalStateException.class,
+ () -> addValue(TypeDescription.createBinary(),
HoodieSchema.create(HoodieSchemaType.BYTES), "not-binary"));
+ assertThrows(IllegalStateException.class,
+ () ->
addValue(TypeDescription.createDecimal().withScale(2).withPrecision(8),
+ HoodieSchema.createDecimal(8, 2), "not-decimal"));
+ assertThrows(org.apache.hudi.exception.HoodieIOException.class,
+ () -> roundTrip(TypeDescription.createVarchar().withMaxLength(3),
stringSchema, "too-long"));
+ }
+
+ @Test
+ public void testUnionMatchesPrimitiveAndBinaryRepresentations() {
+ HoodieSchema schema = HoodieSchema.createUnion(Arrays.asList(
+ HoodieSchema.create(HoodieSchemaType.BOOLEAN),
+ HoodieSchema.create(HoodieSchemaType.INT),
+ HoodieSchema.create(HoodieSchemaType.LONG),
+ HoodieSchema.create(HoodieSchemaType.FLOAT),
+ HoodieSchema.create(HoodieSchemaType.DOUBLE),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.BYTES)));
+ TypeDescription type = TypeDescription.createUnion()
+ .addUnionChild(TypeDescription.createBoolean())
+ .addUnionChild(TypeDescription.createInt())
+ .addUnionChild(TypeDescription.createLong())
+ .addUnionChild(TypeDescription.createFloat())
+ .addUnionChild(TypeDescription.createDouble())
+ .addUnionChild(TypeDescription.createString())
+ .addUnionChild(TypeDescription.createBinary());
+
+ assertTrue((Boolean) roundTrip(type, schema, true));
+ assertEquals(17, roundTrip(type, schema, 17));
+ assertEquals(23L, roundTrip(type, schema, 23L));
+ assertEquals(1.5f, roundTrip(type, schema, 1.5f));
+ assertEquals(2.5d, roundTrip(type, schema, 2.5d));
+ assertEquals("text", roundTrip(type, schema, new Utf8("text")).toString());
+ assertArrayEquals(new byte[] {9, 8}, toByteArray((ByteBuffer) roundTrip(
+ type, schema, new byte[] {9, 8})));
+
+ VectorizedRowBatch batch =
TypeDescription.createStruct().addField("value", type).createRowBatch();
+ assertFalse(AvroOrcUtils.addUnionValue((UnionColumnVector) batch.cols[0],
type.getChildren(), schema,
+ new Object(), 0));
+ }
+
+ @Test
+ public void testTimeAndNullUnionSchemaCreation() {
+ assertEquals(TypeDescription.createInt(),
AvroOrcUtils.createOrcSchema(HoodieSchema.createTimeMillis()));
+ assertEquals(TypeDescription.createLong(),
AvroOrcUtils.createOrcSchema(HoodieSchema.createTimeMicros()));
+ assertEquals(TypeDescription.createUnion(),
+ AvroOrcUtils.createOrcSchema(HoodieSchema.createUnion(Arrays.asList(
+ HoodieSchema.create(HoodieSchemaType.NULL)))));
+ }
+
+ private static Object roundTrip(HoodieSchema schema, Object value) {
+ TypeDescription type = AvroOrcUtils.createOrcSchema(schema);
+ return roundTrip(type, schema, value);
+ }
+
+ private static Object roundTrip(TypeDescription type, HoodieSchema schema,
Object value) {
+ VectorizedRowBatch batch =
TypeDescription.createStruct().addField("value", type).createRowBatch();
+ AvroOrcUtils.addToVector(type, batch.cols[0], schema, value, 0);
+ return AvroOrcUtils.readFromVector(type, batch.cols[0], schema, 0);
+ }
+
+ private static void addValue(TypeDescription type, HoodieSchema schema,
Object value) {
+ VectorizedRowBatch batch =
TypeDescription.createStruct().addField("value", type).createRowBatch();
+ AvroOrcUtils.addToVector(type, batch.cols[0], schema, value, 0);
+ }
+
+ private static byte[] toByteArray(ByteBuffer buffer) {
+ ByteBuffer duplicate = buffer.duplicate();
+ byte[] bytes = new byte[duplicate.remaining()];
+ duplicate.get(bytes);
+ return bytes;
+ }
}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestOrcUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestOrcUtils.java
new file mode 100644
index 000000000000..6c42ce1216c9
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestOrcUtils.java
@@ -0,0 +1,150 @@
+/*
+ * 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.common.util;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.MetadataNotFoundException;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.metadata.HoodieIndexVersion;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.OrcFile;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+
+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;
+
+class TestOrcUtils {
+
+ @Test
+ void testMetadataAndFormatHelpers(@TempDir Path tempDir) throws Exception {
+ StoragePath file = new
StoragePath(tempDir.resolve("metadata.orc").toUri());
+ HoodieStorage storage = HoodieTestUtils.getStorage(file);
+ OrcUtils utils = new OrcUtils();
+ Properties properties = new Properties();
+ properties.setProperty("first", "one");
+ properties.setProperty("second", "two");
+
+ utils.writeMetaFile(storage, file, properties);
+
+ assertEquals(HoodieFileFormat.ORC, utils.getFormat());
+ assertEquals(0, utils.getRowCount(storage, file));
+ assertEquals(Collections.singletonMap("first", "one"),
+ utils.readFooter(storage, true, file, "first"));
+ assertTrue(utils.readFooter(storage, false, file, "missing").isEmpty());
+ assertThrows(MetadataNotFoundException.class,
+ () -> utils.readFooter(storage, true, file, "missing"));
+
assertEquals(HoodieSchemaUtils.getRecordKeySchema().getFields().get(0).name(),
+ utils.readSchema(storage, file).getFields().get(0).name());
+ assertThrows(UnsupportedOperationException.class,
+ () -> utils.readColumnStatsFromMetadata(storage, file,
Collections.emptyList(), HoodieIndexVersion.V1));
+ assertThrows(UnsupportedOperationException.class, () ->
utils.serializeRecordsToLogBlock(
+ storage, Collections.<HoodieRecord>emptyList(), null, null, null,
Collections.emptyMap()));
+ assertThrows(UnsupportedOperationException.class, () ->
utils.serializeRecordsToLogBlock(
+ storage, Collections.<HoodieRecord>emptyIterator(),
HoodieRecord.HoodieRecordType.AVRO,
+ null, null, null, Collections.emptyMap()));
+
+ StoragePath missing = new
StoragePath(tempDir.resolve("missing.orc").toUri());
+ try (ClosableIterator<Pair<HoodieKey, Long>> iterator =
utils.fetchRecordKeysWithPositions(
+ storage, missing, Option.empty(), Option.empty())) {
+ assertFalse(iterator.hasNext());
+ }
+ }
+
+ @Test
+ void testRecordReadingKeyFilteringAndPositions(@TempDir Path tempDir) throws
Exception {
+ StoragePath file = new StoragePath(tempDir.resolve("records.orc").toUri());
+ HoodieStorage storage = HoodieTestUtils.getStorage(file);
+ HoodieSchema schema = HoodieSchemaUtils.getRecordKeySchema();
+ TypeDescription orcSchema = AvroOrcUtils.createOrcSchema(schema);
+ Configuration conf = storage.getConf().unwrapAs(Configuration.class);
+ OrcFile.WriterOptions options = OrcFile.writerOptions(conf)
+ .fileSystem((FileSystem) storage.getFileSystem()).setSchema(orcSchema);
+ try (Writer writer =
OrcFile.createWriter(HadoopFSUtils.convertToHadoopPath(file), options)) {
+ VectorizedRowBatch batch = orcSchema.createRowBatch();
+ BytesColumnVector keys = (BytesColumnVector) batch.cols[0];
+ for (String key : Arrays.asList("key-1", "key-2", "key-3")) {
+ keys.setVal(batch.size++, StringUtils.getUTF8Bytes(key));
+ }
+ writer.addRowBatch(batch);
+ }
+
+ OrcUtils utils = new OrcUtils();
+ assertEquals(3, utils.getRowCount(storage, file));
+ assertEquals(3, utils.readAvroRecords(storage, file).size());
+ assertEquals(3, utils.readAvroRecords(storage, file, schema).size());
+
+ Set<Pair<String, Long>> all = utils.filterRowKeys(storage, file,
Collections.emptySet());
+ assertEquals(3, all.size());
+ Set<Pair<String, Long>> selected = utils.filterRowKeys(
+ storage, file, new HashSet<>(Collections.singletonList("key-2")));
+ assertEquals(Collections.singleton(Pair.of("key-2", 1L)), selected);
+
+ try (ClosableIterator<Pair<HoodieKey, Long>> iterator =
utils.fetchRecordKeysWithPositions(
+ storage, file, Option.empty(), Option.of("partition"))) {
+ assertTrue(iterator.hasNext());
+ Pair<HoodieKey, Long> first = iterator.next();
+ assertEquals("key-1", first.getLeft().getRecordKey());
+ assertEquals("partition", first.getLeft().getPartitionPath());
+ assertEquals(0L, first.getRight());
+ }
+ }
+
+ @Test
+ void testReadSchemaPrefersEmbeddedAvroSchema(@TempDir Path tempDir) throws
Exception {
+ StoragePath file = new StoragePath(tempDir.resolve("schema.orc").toUri());
+ HoodieStorage storage = HoodieTestUtils.getStorage(file);
+ HoodieSchema schema = HoodieSchemaUtils.getRecordKeySchema();
+ TypeDescription orcSchema = AvroOrcUtils.createOrcSchema(schema);
+ Configuration conf = storage.getConf().unwrapAs(Configuration.class);
+ OrcFile.WriterOptions options = OrcFile.writerOptions(conf)
+ .fileSystem((FileSystem) storage.getFileSystem()).setSchema(orcSchema);
+ try (Writer writer =
OrcFile.createWriter(HadoopFSUtils.convertToHadoopPath(file), options)) {
+ writer.addUserMetadata("orc.avro.schema",
ByteBuffer.wrap(StringUtils.getUTF8Bytes(schema.toString())));
+ }
+
+ assertEquals(schema, new OrcUtils().readSchema(storage, file));
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestCachingPath.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestCachingPath.java
new file mode 100644
index 000000000000..d17845900c24
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestCachingPath.java
@@ -0,0 +1,105 @@
+/*
+ * 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.hadoop.fs;
+
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+
+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;
+
+class TestCachingPath {
+
+ @Test
+ void testCachesDerivedValues() {
+ CachingPath path = new
CachingPath("s3://bucket/table/partition/file.parquet");
+
+ String name = path.getName();
+ Path parent = path.getParent();
+ String fullPath = path.toString();
+
+ assertSame(name, path.getName());
+ assertSame(parent, path.getParent());
+ assertSame(fullPath, path.toString());
+ assertEquals("file.parquet", name);
+ assertEquals("s3://bucket/table/partition", parent.toString());
+ }
+
+ @Test
+ void testEveryConstructorCreatesEquivalentPath() {
+ Path parent = new Path("file:///tmp/table");
+ Path child = new Path("partition/file.parquet");
+ URI expected = URI.create("file:/tmp/table/partition/file.parquet");
+
+ assertEquals(expected, new CachingPath(parent.toString(),
child.toString()).toUri());
+ assertEquals(expected, new CachingPath(parent, child.toString()).toUri());
+ assertEquals(expected, new CachingPath(parent.toString(), child).toUri());
+ assertEquals(expected, new CachingPath(parent, child).toUri());
+ assertEquals(expected, new CachingPath(expected).toUri());
+ }
+
+ @Test
+ void testWrapAndSubPath() {
+ CachingPath cached = new CachingPath("s3://bucket/table");
+ assertSame(cached, CachingPath.wrap(cached));
+
+ CachingPath wrapped = CachingPath.wrap(new Path("s3://bucket/table"));
+ assertEquals(cached, wrapped);
+ assertEquals("s3://bucket/table/partition/file.parquet",
+ wrapped.subPath("partition/file.parquet").toString());
+ }
+
+ @Test
+ void testUnsafeConcatenationHandlesSeparatorsAndPreservesUriParts() throws
Exception {
+ Path base = new Path(new URI("s3", "bucket", "/table", "version=1",
"fragment"));
+
+ CachingPath expected = CachingPath.concatPathUnsafe(base, "partition");
+ assertEquals("/table/partition", expected.toUri().getPath());
+ assertEquals("version=1", expected.toUri().getQuery());
+ assertEquals("fragment", expected.toUri().getFragment());
+
+ assertEquals("/table/partition", CachingPath.concatPathUnsafe(new
Path("s3://bucket/table/"), "/partition").toUri().getPath());
+ assertEquals("/table/partition", CachingPath.concatPathUnsafe(new
Path("s3://bucket/table/"), "partition").toUri().getPath());
+ assertEquals("/table/partition", CachingPath.concatPathUnsafe(new
Path("s3://bucket/table"), "/partition").toUri().getPath());
+ assertEquals("/table/partition", CachingPath.concatPathUnsafe(new
Path("s3://bucket/table"), new Path("partition")).toUri().getPath());
+ assertThrows(IllegalStateException.class,
+ () -> CachingPath.concatPathUnsafe(base, new
Path("s3://other/absolute")));
+ }
+
+ @Test
+ void testRelativePathAndSchemeRemoval() {
+ CachingPath relative =
CachingPath.createRelativePathUnsafe("partition%2Fvalue/file.parquet");
+ assertFalse(relative.toUri().isAbsolute());
+ assertEquals("partition%2Fvalue/file.parquet", relative.toString());
+
+ Path absolute = new Path("s3://bucket/table/file.parquet");
+ Path stripped = CachingPath.getPathWithoutSchemeAndAuthority(absolute);
+ assertTrue(stripped instanceof CachingPath);
+ assertEquals("/table/file.parquet", stripped.toString());
+
+ Path alreadyRelative = new Path("table/file.parquet");
+ assertSame(alreadyRelative,
CachingPath.getPathWithoutSchemeAndAuthority(alreadyRelative));
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
index da4e9a7500f4..92e7234eab7d 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
@@ -19,7 +19,12 @@
package org.apache.hudi.hadoop.fs;
+import org.apache.hudi.avro.model.HoodieFSPermission;
+import org.apache.hudi.avro.model.HoodieFileStatus;
+import org.apache.hudi.avro.model.HoodiePath;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
@@ -31,6 +36,9 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FilterFileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
@@ -40,6 +48,11 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopFileStatus;
import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopPath;
@@ -48,7 +61,9 @@ import static
org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePathInfo;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
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.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -255,6 +270,121 @@ public class TestHadoopFSUtils {
convertedPathInfo, path, length, isDirectory, blockReplication,
blockSize, modificationTime);
}
+ @Test
+ public void testStorageConfigurationCopyAndFileSystemOverloads(@TempDir
java.nio.file.Path tempDir) {
+ Configuration conf = new Configuration(false);
+ conf.set("test.key", "before");
+ StorageConfiguration<Configuration> shared =
HadoopFSUtils.getStorageConf(conf);
+ StorageConfiguration<Configuration> copied =
HadoopFSUtils.getStorageConfWithCopy(conf);
+ conf.set("test.key", "after");
+
+ assertEquals("after", shared.getString("test.key").get());
+ assertEquals("before", copied.getString("test.key").get());
+ assertEquals("file", HadoopFSUtils.getFs(tempDir.toUri().toString(),
shared).getScheme());
+ assertEquals("file", HadoopFSUtils.getFs(tempDir.toUri().toString(),
shared, true).getScheme());
+ assertEquals("file", HadoopFSUtils.getFs(new Path(tempDir.toUri()),
shared).getScheme());
+ assertEquals("file", HadoopFSUtils.getFs(new Path(tempDir.toUri()),
shared, true).getScheme());
+ assertEquals("file", HadoopFSUtils.getFs(new StoragePath(tempDir.toUri()),
conf).getScheme());
+ assertEquals("file", HadoopFSUtils.getFs(tempDir.toString(), conf,
true).getScheme());
+ assertInstanceOf(StorageConfiguration.class,
HadoopFSUtils.getStorageConf());
+ }
+
+ @Test
+ public void testAvroPathPermissionAndStatusConversions() {
+ Path path = new Path("s3://bucket/table/file.parquet");
+ HoodiePath hoodiePath = HadoopFSUtils.fromPath(path);
+ assertEquals(path, HadoopFSUtils.toPath(hoodiePath));
+ assertNull(HadoopFSUtils.toPath(null));
+ assertNull(HadoopFSUtils.fromPath(null));
+
+ FsPermission permission = new FsPermission(
+ FsAction.ALL, FsAction.READ_EXECUTE, FsAction.READ, true);
+ HoodieFSPermission hoodiePermission =
HadoopFSUtils.fromFSPermission(permission);
+ assertEquals(permission, HadoopFSUtils.toFSPermission(hoodiePermission));
+ assertNull(HadoopFSUtils.toFSPermission(null));
+ assertNull(HadoopFSUtils.fromFSPermission(null));
+
+ FileStatus status = new FileStatus(123, false, 2, 4096, 1000, 900,
+ permission, "owner", "group", path);
+ HoodieFileStatus converted = HadoopFSUtils.fromFileStatus(status);
+ assertEquals(path, HadoopFSUtils.toPath(converted.getPath()));
+ assertEquals(123, converted.getLength());
+ assertEquals("owner", converted.getOwner());
+ assertEquals("group", converted.getGroup());
+ assertEquals(permission,
HadoopFSUtils.toFSPermission(converted.getPermission()));
+ assertNull(HadoopFSUtils.fromFileStatus(null));
+ }
+
+ @Test
+ public void testStatusLocationsAndFileNameHelpers() {
+ FileStatus fileStatus = new FileStatus(10, false, 1, 128, 1000, new
Path("/table/file.parquet"));
+ StoragePathInfo pathInfo = HadoopFSUtils.convertToStoragePathInfo(
+ fileStatus, new String[] {"host1", "host2"});
+ assertArrayEquals(new String[] {"host1", "host2"},
pathInfo.getLocations());
+
+ assertTrue(HadoopFSUtils.isBaseFile(new Path("fileId_1-0-1_000.parquet")));
+ assertTrue(HadoopFSUtils.isLogFile(new Path(".file_100.log.1_1-0-1")));
+ assertTrue(HadoopFSUtils.isDataFile(new Path("fileId_1-0-1_000.orc")));
+ assertFalse(HadoopFSUtils.isDataFile(new Path("README.md")));
+ assertEquals(new Path("file:///table/partition"),
+ HadoopFSUtils.constructAbsolutePathInHadoopPath("file:///table",
"partition"));
+ }
+
+ @Test
+ public void testDfsFullPartitionPath() throws IOException {
+ try (FileSystem fs = FileSystem.newInstanceLocal(new Configuration())) {
+ assertEquals(fs.getUri() + "/tmp/table",
+ HadoopFSUtils.getDFSFullPartitionPath(fs, new Path("/tmp/table")));
+ }
+ }
+
+ @Test
+ public void testFileNameAndRelativePathDelegates() {
+ assertEquals("partition", HadoopFSUtils.getRelativePartitionPath(
+ new Path("/table"), new Path("/table/partition")));
+ Path logPath = new Path("/table/.file-id_001.log.1_1-0-1");
+ assertEquals("file-id", HadoopFSUtils.getFileIdFromLogPath(logPath));
+ assertEquals("001", HadoopFSUtils.getDeltaCommitTimeFromLogPath(logPath));
+ }
+
+ @Test
+ public void testRecoverLeaseStopsAfterSuccess() throws Exception {
+ AtomicInteger attempts = new AtomicInteger();
+ DistributedFileSystem fs = new DistributedFileSystem() {
+ @Override
+ public boolean recoverLease(Path path) {
+ attempts.incrementAndGet();
+ return true;
+ }
+ };
+
+ assertTrue(HadoopFSUtils.recoverDFSFileLease(fs, new Path("/table/file")));
+ assertEquals(1, attempts.get());
+ }
+
+ @Test
+ public void testParallelFileProcessingAndStatusAtLevel(@TempDir
java.nio.file.Path tempDir) throws Exception {
+ Configuration conf = new Configuration();
+ HoodieLocalEngineContext context = new
HoodieLocalEngineContext(HadoopFSUtils.getStorageConf(conf));
+ try (FileSystem fs = FileSystem.newInstanceLocal(conf)) {
+ Map<String, Integer> lengths = HadoopFSUtils.parallelizeFilesProcess(
+ context, fs, 2, pair -> pair.getKey().length(), Arrays.asList("a",
"longer"));
+ assertEquals(1, lengths.get("a"));
+ assertEquals(6, lengths.get("longer"));
+ assertTrue(HadoopFSUtils.parallelizeFilesProcess(
+ context, fs, 2, pair -> pair.getKey().length(),
Collections.<String>emptyList()).isEmpty());
+
+ Path root = new Path(tempDir.resolve("root").toUri());
+ Path secondLevel = new Path(new Path(root, "first"), "second");
+ assertTrue(fs.mkdirs(secondLevel));
+ Path leaf = new Path(secondLevel, "leaf.txt");
+ assertTrue(fs.createNewFile(leaf));
+ List<FileStatus> statuses = HadoopFSUtils.getFileStatusAtLevel(context,
fs, root, 2, 2);
+ assertEquals(Collections.singletonList(leaf),
+
statuses.stream().map(FileStatus::getPath).collect(java.util.stream.Collectors.toList()));
+ }
+ }
+
private void assertFileStatus(FileStatus fileStatus,
String path,
long length,
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHoodieRetryWrapperFileSystem.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHoodieRetryWrapperFileSystem.java
new file mode 100644
index 000000000000..3a091f283a2d
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHoodieRetryWrapperFileSystem.java
@@ -0,0 +1,158 @@
+/*
+ * 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.hadoop.fs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FilterFileSystem;
+import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.EnumSet;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestHoodieRetryWrapperFileSystem {
+
+ @Test
+ void testFileOperationsDelegateToWrappedFileSystem(@TempDir
java.nio.file.Path tempDir) throws IOException {
+ try (FileSystem local = FileSystem.newInstanceLocal(new Configuration())) {
+ HoodieRetryWrapperFileSystem fs = new
HoodieRetryWrapperFileSystem(local, 0, 0, 0, "");
+ Path directory = new Path(tempDir.toUri());
+ Path file = new Path(directory, "data.bin");
+
+ assertEquals(local.getUri(), fs.getUri());
+ assertEquals(local.getConf(), fs.getConf());
+ assertEquals("file", fs.getScheme());
+ assertEquals(local.getDefaultReplication(), fs.getDefaultReplication());
+ assertEquals(local.getDefaultReplication(file),
fs.getDefaultReplication(file));
+ fs.setWorkingDirectory(directory);
+ assertEquals(directory, fs.getWorkingDirectory());
+ assertTrue(fs.mkdirs(directory, FsPermission.getDirDefault()));
+
+ try (FSDataOutputStream output = fs.create(file)) {
+ output.write(new byte[] {1, 2, 3});
+ }
+ assertTrue(fs.exists(file));
+ assertEquals(3, fs.getFileStatus(file).getLen());
+ fs.open(file).close();
+ fs.open(file, 128).close();
+
+ Path[] files = new Path[] {file};
+ assertEquals(1, fs.listStatus(directory).length);
+ assertEquals(1, fs.listStatus(directory, path -> true).length);
+ assertEquals(1, fs.listStatus(files).length);
+ assertEquals(1, fs.listStatus(files, path -> true).length);
+ assertEquals(1, fs.globStatus(new Path(directory, "*.bin")).length);
+ assertEquals(1, fs.globStatus(new Path(directory, "*.bin"), path ->
true).length);
+ assertTrue(fs.listLocatedStatus(directory).hasNext());
+ assertTrue(fs.listFiles(directory, true).hasNext());
+
+ Path renamed = new Path(directory, "renamed.bin");
+ assertTrue(fs.rename(file, renamed));
+ assertTrue(fs.delete(renamed, false));
+ assertFalse(fs.exists(renamed));
+ assertTrue(fs.createNewFile(new Path(directory, "empty")));
+ }
+ }
+
+ @Test
+ void testCreateOverloadsDelegate(@TempDir java.nio.file.Path tempDir) throws
IOException {
+ try (FileSystem local = FileSystem.newInstanceLocal(new Configuration())) {
+ HoodieRetryWrapperFileSystem fs = new
HoodieRetryWrapperFileSystem(local, 0, 0, 0, "");
+ Path directory = new Path(tempDir.toUri());
+ FsPermission permission = FsPermission.getFileDefault();
+ short replication = local.getDefaultReplication(directory);
+ long blockSize = local.getDefaultBlockSize(directory);
+ AtomicInteger suffix = new AtomicInteger();
+
+ close(fs.create(nextPath(directory, suffix), true));
+ close(fs.create(nextPath(directory, suffix), () -> { }));
+ close(fs.create(nextPath(directory, suffix), replication));
+ close(fs.create(nextPath(directory, suffix), replication, () -> { }));
+ close(fs.create(nextPath(directory, suffix), true, 4096));
+ close(fs.create(nextPath(directory, suffix), true, 4096, () -> { }));
+ close(fs.create(nextPath(directory, suffix), true, 4096, replication,
blockSize));
+ close(fs.create(nextPath(directory, suffix), true, 4096, replication,
blockSize, () -> { }));
+ close(fs.create(nextPath(directory, suffix), permission, true, 4096,
replication, blockSize, () -> { }));
+ close(fs.create(nextPath(directory, suffix), permission,
EnumSet.of(CreateFlag.CREATE),
+ 4096, replication, blockSize, () -> { }));
+ close(fs.create(nextPath(directory, suffix), permission,
EnumSet.of(CreateFlag.CREATE),
+ 4096, replication, blockSize, () -> { },
Options.ChecksumOpt.createDisabled()));
+ }
+ }
+
+ @Test
+ void testDeleteRetriesWhenDelegateReportsExistingFile() throws IOException {
+ FlakyDeleteFileSystem delegate = new
FlakyDeleteFileSystem(FileSystem.newInstanceLocal(new Configuration()));
+ try {
+ HoodieRetryWrapperFileSystem fs = new
HoodieRetryWrapperFileSystem(delegate, 0, 2, 0, "");
+
+ assertTrue(fs.delete(new Path("file:///eventually-deleted")));
+ assertEquals(3, delegate.deleteAttempts);
+ assertEquals(2, delegate.existsChecks);
+ } finally {
+ delegate.close();
+ }
+ }
+
+ private static Path nextPath(Path directory, AtomicInteger suffix) {
+ return new Path(directory, "create-" + suffix.incrementAndGet());
+ }
+
+ private static void close(FSDataOutputStream stream) throws IOException {
+ stream.close();
+ }
+
+ private static class FlakyDeleteFileSystem extends FilterFileSystem {
+ private int deleteAttempts;
+ private int existsChecks;
+
+ private FlakyDeleteFileSystem(FileSystem delegate) {
+ super(delegate);
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) {
+ return ++deleteAttempts >= 3;
+ }
+
+ @Override
+ public boolean exists(Path path) {
+ existsChecks++;
+ return true;
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ return fs.getFileStatus(path);
+ }
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHoodieWrapperFileSystemOperations.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHoodieWrapperFileSystemOperations.java
new file mode 100644
index 000000000000..b9f6864464ce
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHoodieWrapperFileSystemOperations.java
@@ -0,0 +1,476 @@
+/*
+ * 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.hadoop.fs;
+
+import org.apache.hudi.common.fs.ConsistencyGuard;
+import org.apache.hudi.common.fs.NoOpConsistencyGuard;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.XAttrSetFlag;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.Credentials;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+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.assertSame;
+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.verify;
+import static org.mockito.Mockito.when;
+
+class TestHoodieWrapperFileSystemOperations {
+
+ @Test
+ void testPathTranslationAndInitialization(@TempDir java.nio.file.Path
tempDir) throws Exception {
+ Configuration conf = new Configuration();
+ try (FileSystem local = FileSystem.newInstanceLocal(conf)) {
+ HoodieWrapperFileSystem fs = new HoodieWrapperFileSystem(local, new
NoOpConsistencyGuard());
+ Path path = new Path(tempDir.resolve("file.parquet").toUri());
+
+ assertEquals("hoodie-file",
HoodieWrapperFileSystem.getHoodieScheme("file"));
+ assertThrows(IllegalArgumentException.class, () ->
HoodieWrapperFileSystem.getHoodieScheme("unknown"));
+ assertEquals("hoodie-file", HoodieWrapperFileSystem.convertToHoodiePath(
+ new StoragePath(path.toUri()), conf).toUri().getScheme());
+
+ Path converted = HoodieWrapperFileSystem.convertPathWithScheme(
+ new Path(new URI("file", null, path.toUri().getPath(), "query=1",
"fragment")), "hoodie-file");
+ assertTrue(converted instanceof CachingPath);
+ assertEquals("hoodie-file", converted.toUri().getScheme());
+ assertEquals("query=1", converted.toUri().getQuery());
+ assertEquals("fragment", converted.toUri().getFragment());
+
+ assertEquals("file", fs.getScheme());
+ assertEquals("hoodie-file",
fs.convertToHoodiePath(path).toUri().getScheme());
+ assertSame(local, fs.getFileSystem());
+ assertEquals(local.getUri(), fs.getUri());
+ assertEquals(local.getConf(), fs.getConf());
+ assertEquals(local.hashCode(), fs.hashCode());
+ assertTrue(fs.equals(local));
+ assertEquals(local.toString(), fs.toString());
+ fs.setConf(new Configuration());
+ assertSame(local.getConf(), fs.getConf());
+ fs.close();
+
+ HoodieWrapperFileSystem initialized = new HoodieWrapperFileSystem();
+ initialized.initialize(URI.create("hoodie-file:" +
tempDir.toUri().getPath()), conf);
+ assertEquals("file", initialized.getScheme());
+ assertEquals("file", initialized.getFileSystem().getScheme());
+
+ HoodieWrapperFileSystem initializedWithoutPrefix = new
HoodieWrapperFileSystem();
+ initializedWithoutPrefix.initialize(tempDir.toUri(), conf);
+ assertEquals("file", initializedWithoutPrefix.getScheme());
+ }
+ }
+
+ @Test
+ void testLocalFileOperationsAndStreamAccounting(@TempDir java.nio.file.Path
tempDir) throws IOException {
+ try (FileSystem local = FileSystem.newInstanceLocal(new Configuration())) {
+ HoodieWrapperFileSystem fs = new HoodieWrapperFileSystem(local, new
NoOpConsistencyGuard());
+ Path directory = new Path(tempDir.resolve("table").toUri());
+ Path file = new Path(directory, "data.bin");
+
+ assertTrue(fs.mkdirs(directory));
+ assertTrue(fs.mkdirs(directory, FsPermission.getDirDefault()));
+ fs.setWorkingDirectory(directory);
+ assertEquals("hoodie-file",
fs.getWorkingDirectory().toUri().getScheme());
+ assertEquals("hoodie-file", fs.getHomeDirectory().toUri().getScheme());
+
+ try (FSDataOutputStream output = fs.create(file)) {
+ output.write(new byte[] {1, 2, 3, 4}, 0, 4);
+ assertEquals(4, fs.getBytesWritten(file));
+ }
+ assertThrows(IllegalArgumentException.class, () ->
fs.getBytesWritten(file));
+
+ try (FSDataInputStream input = fs.open(file)) {
+ assertEquals(1, input.read());
+ }
+ try (FSDataInputStream input = fs.open(file, 128)) {
+ assertEquals(1, input.read());
+ }
+
+ assertTrue(fs.exists(file));
+ assertTrue(fs.isFile(file));
+ assertFalse(fs.isDirectory(file));
+ assertEquals(4, fs.getLength(file));
+ assertEquals(4, fs.getFileStatus(file).getLen());
+ assertEquals(4, fs.getContentSummary(file).getLength());
+ assertEquals(1, fs.listStatus(directory).length);
+ assertEquals(1, fs.listStatus(directory, candidate -> true).length);
+ assertEquals(1, fs.listStatus(new Path[] {file}).length);
+ assertEquals(1, fs.listStatus(new Path[] {file}, candidate ->
true).length);
+ assertEquals(0, fs.listStatus(new Path[0]).length);
+ assertEquals(0, fs.listStatus(new Path[0], candidate -> true).length);
+ assertEquals(1, fs.globStatus(new Path(directory, "*.bin")).length);
+ assertEquals(1, fs.globStatus(new Path(directory, "*.bin"), candidate ->
true).length);
+ assertTrue(fs.listLocatedStatus(directory).hasNext());
+ assertTrue(fs.listFiles(directory, true).hasNext());
+
+ FileStatus status = fs.getFileStatus(file);
+ assertNotNull(fs.getFileBlockLocations(status, 0, status.getLen()));
+ assertNotNull(fs.getFileBlockLocations(file, 0, status.getLen()));
+ assertNotNull(fs.getServerDefaults());
+ assertNotNull(fs.getServerDefaults(file));
+ assertNotNull(fs.getStatus());
+ assertNotNull(fs.getStatus(file));
+ assertTrue(fs.getBlockSize(file) > 0);
+ assertTrue(fs.getDefaultBlockSize() > 0);
+ assertTrue(fs.getDefaultBlockSize(file) > 0);
+ assertTrue(fs.getDefaultReplication() > 0);
+ assertTrue(fs.getDefaultReplication(file) > 0);
+ assertTrue(fs.getReplication(file) > 0);
+ assertTrue(fs.setReplication(file, fs.getReplication(file)));
+ fs.access(file, FsAction.READ);
+ fs.setPermission(file, FsPermission.getFileDefault());
+ fs.setTimes(file, System.currentTimeMillis(), -1);
+ fs.setVerifyChecksum(true);
+ fs.setWriteChecksum(true);
+ assertEquals(local.getFileChecksum(file), fs.getFileChecksum(file));
+
+ assertEquals("hoodie-file", fs.makeQualified(file).toUri().getScheme());
+ assertEquals("hoodie-file", fs.resolvePath(file).toUri().getScheme());
+ assertEquals(local.getCanonicalServiceName(),
fs.getCanonicalServiceName());
+ assertEquals(local.getName(), fs.getName());
+ assertEquals(local.supportsSymlinks(), fs.supportsSymlinks());
+ assertArrayEquals(local.getChildFileSystems(), fs.getChildFileSystems());
+
+ Path renamed = new Path(directory, "renamed.bin");
+ assertTrue(fs.rename(file, renamed));
+ assertTrue(fs.delete(renamed));
+ assertFalse(fs.exists(renamed));
+ assertTrue(fs.createNewFile(new Path(directory, "empty")));
+ }
+ }
+
+ @Test
+ void testCreateOverloads(@TempDir java.nio.file.Path tempDir) throws
IOException {
+ try (FileSystem local = FileSystem.newInstanceLocal(new Configuration())) {
+ HoodieWrapperFileSystem fs = new HoodieWrapperFileSystem(local, new
NoOpConsistencyGuard());
+ Path directory = new Path(tempDir.toUri());
+ FsPermission permission = FsPermission.getFileDefault();
+ short replication = local.getDefaultReplication(directory);
+ long blockSize = local.getDefaultBlockSize(directory);
+ AtomicInteger suffix = new AtomicInteger();
+
+ close(fs.create(nextPath(directory, suffix), true));
+ close(fs.create(nextPath(directory, suffix), () -> { }));
+ close(fs.create(nextPath(directory, suffix), replication));
+ close(fs.create(nextPath(directory, suffix), replication, () -> { }));
+ close(fs.create(nextPath(directory, suffix), true, 4096));
+ close(fs.create(nextPath(directory, suffix), true, 4096, () -> { }));
+ close(fs.create(nextPath(directory, suffix), true, 4096, replication,
blockSize));
+ close(fs.create(nextPath(directory, suffix), true, 4096, replication,
blockSize, () -> { }));
+ close(fs.create(nextPath(directory, suffix), permission, true, 4096,
replication, blockSize, () -> { }));
+ close(fs.create(nextPath(directory, suffix), permission,
EnumSet.of(CreateFlag.CREATE),
+ 4096, replication, blockSize, () -> { }));
+ close(fs.create(nextPath(directory, suffix), permission,
EnumSet.of(CreateFlag.CREATE),
+ 4096, replication, blockSize, () -> { },
Options.ChecksumOpt.createDisabled()));
+
+ Path nonRecursive = nextPath(directory, suffix);
+ close(fs.createNonRecursive(nonRecursive, true, 4096, replication,
blockSize, () -> { }));
+ close(fs.createNonRecursive(nextPath(directory, suffix), permission,
true,
+ 4096, replication, blockSize, () -> { }));
+ close(fs.createNonRecursive(nextPath(directory, suffix), permission,
EnumSet.of(CreateFlag.CREATE),
+ 4096, replication, blockSize, () -> { }));
+ }
+ }
+
+ @Test
+ void testDataAndMetadataMetricsAreSeparated(@TempDir java.nio.file.Path
tempDir) throws IOException {
+ Registry dataRegistry = Registry.getRegistry("wrapper-coverage-data-" +
System.nanoTime());
+ Registry metaRegistry = Registry.getRegistry("wrapper-coverage-meta-" +
System.nanoTime());
+ HoodieWrapperFileSystem.setMetricsRegistry(dataRegistry, metaRegistry);
+ try (FileSystem local = FileSystem.newInstanceLocal(new Configuration())) {
+ HoodieWrapperFileSystem fs = new HoodieWrapperFileSystem(local, new
NoOpConsistencyGuard());
+ Path dataDirectory = new Path(tempDir.resolve("data").toUri());
+ Path metaDirectory = new Path(tempDir.resolve(".hoodie").toUri());
+
+ assertTrue(fs.mkdirs(dataDirectory));
+ assertTrue(fs.mkdirs(metaDirectory));
+ assertEquals(1L, dataRegistry.getAllCounts().get("mkdirs"));
+ assertEquals(1L, metaRegistry.getAllCounts().get("mkdirs"));
+
+ HoodieWrapperFileSystem.executeFuncWithTimeAndByteMetrics(
+ "write", dataDirectory, 17, () -> true);
+ assertEquals(17L, dataRegistry.getAllCounts().get("write.totalBytes"));
+ assertEquals(1L, dataRegistry.getAllCounts().get("write"));
+ } finally {
+ HoodieWrapperFileSystem.setMetricsRegistry(null, null);
+ }
+ }
+
+ @Test
+ void testOptionalDelegatedFileSystemApis() throws IOException {
+ FileSystem delegate = mock(FileSystem.class);
+ when(delegate.getUri()).thenReturn(URI.create("file:///"));
+ when(delegate.getScheme()).thenReturn("file");
+ when(delegate.getUsed()).thenReturn(42L);
+
+ HoodieWrapperFileSystem fs = wrapper(delegate, new NoOpConsistencyGuard());
+ Path directory = new Path("hoodie-file:///optional");
+ Path file = new Path(directory, "data.bin");
+ Path link = new Path(directory, "data.link");
+ Path deleteOnExit = new Path(directory, "delete-on-exit");
+ Path defaultDirectory = new Path("file:///optional");
+ Path defaultFile = new Path(defaultDirectory, "data.bin");
+ Path defaultLink = new Path(defaultDirectory, "data.link");
+ Path defaultDeleteOnExit = new Path(defaultDirectory, "delete-on-exit");
+ Credentials credentials = new Credentials();
+ List<AclEntry> aclEntries = Collections.emptyList();
+ byte[] attribute = new byte[] {1, 2};
+ EnumSet<XAttrSetFlag> xAttrFlags = EnumSet.of(XAttrSetFlag.REPLACE);
+ List<String> xAttrNames = Collections.singletonList("user.hudi");
+
+ when(delegate.getLinkTarget(defaultLink)).thenReturn(defaultFile);
+ when(delegate.createSnapshot(defaultDirectory,
"snapshot")).thenReturn(defaultDirectory);
+ when(delegate.deleteOnExit(defaultDeleteOnExit)).thenReturn(true);
+ when(delegate.cancelDeleteOnExit(defaultDeleteOnExit)).thenReturn(true);
+
+ fs.getDelegationToken("renewer");
+ fs.addDelegationTokens("renewer", credentials);
+ fs.listCorruptFileBlocks(directory);
+ assertEquals(42L, fs.getUsed());
+ fs.getFileChecksum(file, 1);
+ fs.setOwner(file, "owner", null);
+ fs.createSymlink(file, link, false);
+ fs.getFileLinkStatus(link);
+ assertEquals("hoodie-file", fs.getLinkTarget(link).toUri().getScheme());
+ assertEquals("hoodie-file", fs.createSnapshot(directory,
"snapshot").toUri().getScheme());
+ fs.renameSnapshot(directory, "snapshot", "renamed");
+ fs.deleteSnapshot(directory, "renamed");
+ fs.modifyAclEntries(file, aclEntries);
+ fs.removeAclEntries(file, aclEntries);
+ fs.removeDefaultAcl(file);
+ fs.removeAcl(file);
+ fs.setAcl(file, aclEntries);
+ fs.getAclStatus(file);
+ fs.setXAttr(file, "user.hudi", attribute);
+ fs.setXAttr(file, "user.hudi", attribute, xAttrFlags);
+ fs.getXAttr(file, "user.hudi");
+ fs.getXAttrs(file);
+ fs.getXAttrs(file, xAttrNames);
+ fs.listXAttrs(file);
+ fs.removeXAttr(file, "user.hudi");
+ assertTrue(fs.deleteOnExit(deleteOnExit));
+ assertTrue(fs.cancelDeleteOnExit(deleteOnExit));
+
+ verify(delegate).getDelegationToken("renewer");
+ verify(delegate).addDelegationTokens("renewer", credentials);
+ verify(delegate).listCorruptFileBlocks(defaultDirectory);
+ verify(delegate).getUsed();
+ verify(delegate).getFileChecksum(defaultFile, 1);
+ verify(delegate).setOwner(defaultFile, "owner", null);
+ verify(delegate).createSymlink(defaultFile, defaultLink, false);
+ verify(delegate).getFileLinkStatus(defaultLink);
+ verify(delegate).getLinkTarget(defaultLink);
+ verify(delegate).createSnapshot(defaultDirectory, "snapshot");
+ verify(delegate).renameSnapshot(defaultDirectory, "snapshot", "renamed");
+ verify(delegate).deleteSnapshot(defaultDirectory, "renamed");
+ verify(delegate).modifyAclEntries(defaultFile, aclEntries);
+ verify(delegate).removeAclEntries(defaultFile, aclEntries);
+ verify(delegate).removeDefaultAcl(defaultFile);
+ verify(delegate).removeAcl(defaultFile);
+ verify(delegate).setAcl(defaultFile, aclEntries);
+ verify(delegate).getAclStatus(defaultFile);
+ verify(delegate).setXAttr(defaultFile, "user.hudi", attribute);
+ verify(delegate).setXAttr(defaultFile, "user.hudi", attribute, xAttrFlags);
+ verify(delegate).getXAttr(defaultFile, "user.hudi");
+ verify(delegate).getXAttrs(defaultFile);
+ verify(delegate).getXAttrs(defaultFile, xAttrNames);
+ verify(delegate).listXAttrs(defaultFile);
+ verify(delegate).removeXAttr(defaultFile, "user.hudi");
+ verify(delegate).deleteOnExit(defaultDeleteOnExit);
+ verify(delegate).cancelDeleteOnExit(defaultDeleteOnExit);
+ }
+
+ @Test
+ void testLocalCopyAndMoveOverloads(@TempDir java.nio.file.Path tempDir)
throws Exception {
+ try (FileSystem local = FileSystem.newInstanceLocal(new Configuration())) {
+ HoodieWrapperFileSystem fs = new HoodieWrapperFileSystem(local, new
NoOpConsistencyGuard());
+ Path destinationDirectory = new
Path(tempDir.resolve("destinations").toUri());
+ assertTrue(fs.mkdirs(destinationDirectory));
+
+ Path copied = new Path(destinationDirectory, "copied");
+ fs.copyFromLocalFile(localFile(tempDir, "copy-source", 1), copied);
+ assertTrue(fs.exists(copied));
+
+ Path moved = new Path(destinationDirectory, "moved");
+ fs.moveFromLocalFile(localFile(tempDir, "move-source", 2), moved);
+ assertTrue(fs.exists(moved));
+
+ Path copiedWithFlag = new Path(destinationDirectory, "copied-with-flag");
+ fs.copyFromLocalFile(false, localFile(tempDir, "copy-flag-source", 3),
copiedWithFlag);
+ assertTrue(fs.exists(copiedWithFlag));
+
+ Path copiedWithFlags = new Path(destinationDirectory,
"copied-with-flags");
+ fs.copyFromLocalFile(false, true, localFile(tempDir,
"copy-flags-source", 4), copiedWithFlags);
+ assertTrue(fs.exists(copiedWithFlags));
+
+ Path arrayDestination = new Path(destinationDirectory, "array-copy");
+ assertTrue(fs.mkdirs(arrayDestination));
+ fs.copyFromLocalFile(false, true,
+ new Path[] {localFile(tempDir, "array-source-1", 5),
localFile(tempDir, "array-source-2", 6)},
+ arrayDestination);
+ assertEquals(2, fs.listStatus(arrayDestination).length);
+
+ Path moveArrayDestination = new Path(destinationDirectory, "array-move");
+ assertTrue(fs.mkdirs(moveArrayDestination));
+ fs.moveFromLocalFile(
+ new Path[] {localFile(tempDir, "move-array-source-1", 7),
localFile(tempDir, "move-array-source-2", 8)},
+ moveArrayDestination);
+ assertEquals(2, fs.listStatus(moveArrayDestination).length);
+
+ Path localCopy = new Path(tempDir.resolve("local-copy").toUri());
+ fs.copyToLocalFile(copied, localCopy);
+ assertTrue(local.exists(localCopy));
+ Path localCopyWithFlag = new
Path(tempDir.resolve("local-copy-flag").toUri());
+ fs.copyToLocalFile(false, copied, localCopyWithFlag);
+ assertTrue(local.exists(localCopyWithFlag));
+ Path localRawCopy = new Path(tempDir.resolve("local-raw-copy").toUri());
+ fs.copyToLocalFile(false, copied, localRawCopy, true);
+ assertTrue(local.exists(localRawCopy));
+
+ Path localMove = new Path(tempDir.resolve("local-move").toUri());
+ fs.moveToLocalFile(copiedWithFlag, localMove);
+ assertTrue(local.exists(localMove));
+ }
+ }
+
+ @Test
+ void testConsistencyGuardTimeoutBranches(@TempDir java.nio.file.Path
tempDir) throws Exception {
+ try (FileSystem local = FileSystem.newInstanceLocal(new Configuration())) {
+ Path root = new Path(tempDir.toUri());
+
+ HoodieWrapperFileSystem mkdirFs = wrapper(local, new
FailingConsistencyGuard(1, 0));
+ assertThrows(HoodieException.class, () -> mkdirFs.mkdirs(new Path(root,
"mkdir-timeout")));
+ HoodieWrapperFileSystem permissionMkdirFs = wrapper(local, new
FailingConsistencyGuard(1, 0));
+ assertThrows(HoodieException.class, () -> permissionMkdirFs.mkdirs(
+ new Path(root, "permission-mkdir-timeout"),
FsPermission.getDirDefault()));
+
+ HoodieWrapperFileSystem createFs = wrapper(local, new
FailingConsistencyGuard(1, 0));
+ assertThrows(HoodieException.class, () -> createFs.createNewFile(new
Path(root, "create-timeout")));
+
+ Path statusFile = new Path(root, "status-timeout");
+ assertTrue(local.createNewFile(statusFile));
+ HoodieWrapperFileSystem statusFs = wrapper(local, new
FailingConsistencyGuard(1, 0));
+ assertEquals(0, statusFs.getFileStatus(statusFile).getLen());
+
+ Path deleteFile = new Path(root, "delete-timeout");
+ assertTrue(local.createNewFile(deleteFile));
+ HoodieWrapperFileSystem deleteFs = wrapper(local, new
FailingConsistencyGuard(0, 1));
+ assertThrows(HoodieException.class, () -> deleteFs.delete(deleteFile,
false));
+
+ Path renameBefore = new Path(root, "rename-before");
+ assertTrue(local.createNewFile(renameBefore));
+ HoodieWrapperFileSystem renameBeforeFs = wrapper(local, new
FailingConsistencyGuard(1, 0));
+ assertThrows(HoodieException.class,
+ () -> renameBeforeFs.rename(renameBefore, new Path(root,
"rename-before-destination")));
+
+ Path renameDestination = new Path(root, "rename-destination");
+ assertTrue(local.createNewFile(renameDestination));
+ HoodieWrapperFileSystem renameDestinationFs = wrapper(local, new
FailingConsistencyGuard(2, 0));
+ assertThrows(HoodieException.class,
+ () -> renameDestinationFs.rename(renameDestination, new Path(root,
"rename-destination-result")));
+
+ Path renameDisappear = new Path(root, "rename-disappear");
+ assertTrue(local.createNewFile(renameDisappear));
+ HoodieWrapperFileSystem renameDisappearFs = wrapper(local, new
FailingConsistencyGuard(0, 1));
+ assertThrows(HoodieException.class,
+ () -> renameDisappearFs.rename(renameDisappear, new Path(root,
"rename-disappear-result")));
+ }
+ }
+
+ private static Path nextPath(Path directory, AtomicInteger suffix) {
+ return new Path(directory, "create-" + suffix.incrementAndGet());
+ }
+
+ private static void close(FSDataOutputStream stream) throws IOException {
+ stream.close();
+ }
+
+ private static HoodieWrapperFileSystem wrapper(FileSystem fileSystem,
ConsistencyGuard consistencyGuard) {
+ return new HoodieWrapperFileSystem(fileSystem, consistencyGuard);
+ }
+
+ private static Path localFile(java.nio.file.Path tempDir, String name, int
value) throws IOException {
+ java.nio.file.Path file = tempDir.resolve(name);
+ Files.write(file, new byte[] {(byte) value});
+ return new Path(file.toUri());
+ }
+
+ private static final class FailingConsistencyGuard implements
ConsistencyGuard {
+ private final int failAppearCall;
+ private final int failDisappearCall;
+ private int appearCalls;
+ private int disappearCalls;
+
+ private FailingConsistencyGuard(int failAppearCall, int failDisappearCall)
{
+ this.failAppearCall = failAppearCall;
+ this.failDisappearCall = failDisappearCall;
+ }
+
+ @Override
+ public void waitTillFileAppears(StoragePath filePath) throws
TimeoutException {
+ if (++appearCalls == failAppearCall) {
+ throw new TimeoutException(filePath.toString());
+ }
+ }
+
+ @Override
+ public void waitTillFileDisappears(StoragePath filePath) throws
TimeoutException {
+ if (++disappearCalls == failDisappearCall) {
+ throw new TimeoutException(filePath.toString());
+ }
+ }
+
+ @Override
+ public void waitTillAllFilesAppear(String dirPath, List<String> files) {
+ }
+
+ @Override
+ public void waitTillAllFilesDisappear(String dirPath, List<String> files) {
+ }
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
index 872a98a16460..f522c21d1caf 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
@@ -19,6 +19,7 @@
package org.apache.hudi.io.storage.hadoop;
+import org.apache.hudi.common.avro.VariantShreddingProvider;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
@@ -27,12 +28,20 @@ import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.storage.HoodieStorage;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.nio.ByteBuffer;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.Collections;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -72,4 +81,85 @@ class TestHoodieVariantReconstruction {
assertNull(HoodieVariantReconstruction.create(schema, schema,
storageWithReadingShredded(tmp, false)));
}
+
+ @Test
+ void returnsNullForNonRecordSchemas(@TempDir Path tmp) {
+ HoodieStorage storage = storageWithReadingShredded(tmp, true);
+ assertNull(HoodieVariantReconstruction.create(
+ HoodieSchema.create(HoodieSchemaType.STRING),
recordWithVariant(HoodieSchema.createVariant()), storage));
+ assertNull(HoodieVariantReconstruction.create(
+ recordWithVariant(HoodieSchema.createVariant()),
HoodieSchema.create(HoodieSchemaType.STRING), storage));
+ }
+
+ @Test
+ void failsFastWhenProviderIsUnavailable(@TempDir Path tmp) {
+ HoodieSchema fileSchema = recordWithVariant(
+
HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.INT)));
+ HoodieSchema requestedSchema =
recordWithVariant(HoodieSchema.createVariant());
+
+ HoodieException exception = assertThrows(HoodieException.class, () ->
+ HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ storageWithReadingShredded(tmp, true)));
+ assertTrue(exception.getMessage().contains("no VariantShreddingProvider is
available"));
+ }
+
+ @Test
+ void reconstructsTargetFieldsAndPassesThroughOtherValues(@TempDir Path tmp) {
+ HoodieSchema fileSchema = recordWithIdAndVariant(
+
HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.INT)));
+ HoodieSchema requestedSchema =
recordWithIdAndVariant(HoodieSchema.createVariant());
+ HoodieStorage storage = storageWithReadingShredded(tmp, true);
+
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
+ TestVariantShreddingProvider.class.getName());
+
+ HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(
+ fileSchema, requestedSchema, storage);
+ assertNotNull(reconstruction);
+
assertTrue(reconstruction.intermediateSchema().getField("v").get().schema().getNonNullType()
+ instanceof HoodieSchema.Variant);
+ assertTrue(((HoodieSchema.Variant)
reconstruction.intermediateSchema().getField("v").get()
+ .schema().getNonNullType()).isShredded());
+
+ GenericRecord shredded = new GenericData.Record(
+
reconstruction.intermediateSchema().getField("v").get().schema().getNonNullType().toAvroSchema());
+ shredded.put("metadata", ByteBuffer.wrap(new byte[] {1}));
+ shredded.put("value", null);
+ shredded.put("typed_value", 42);
+ GenericRecord input = new
GenericData.Record(reconstruction.intermediateSchema().toAvroSchema());
+ input.put("id", "record-1");
+ input.put("v", shredded);
+
+ IndexedRecord output = reconstruction.reconstruct(input);
+ assertEquals("record-1", output.get(0).toString());
+ GenericRecord variant = (GenericRecord) output.get(1);
+ assertEquals(ByteBuffer.wrap(new byte[] {1}), variant.get("metadata"));
+ assertEquals(ByteBuffer.wrap(new byte[] {42}), variant.get("value"));
+
+ input.put("v", null);
+ assertNull(reconstruction.reconstruct(input).get(1));
+ }
+
+ private static HoodieSchema recordWithIdAndVariant(HoodieSchema
variantSchema) {
+ return HoodieSchema.createRecord("test_record", "org.apache.hudi.test",
null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("v", variantSchema)));
+ }
+
+ public static class TestVariantShreddingProvider implements
VariantShreddingProvider {
+ @Override
+ public GenericRecord shredVariantRecord(
+ GenericRecord unshreddedVariant, Schema shreddedSchema,
HoodieSchema.Variant variantSchema) {
+ throw new UnsupportedOperationException("Not used by the reconstruction
test");
+ }
+
+ @Override
+ public GenericRecord rebuildVariantRecord(
+ GenericRecord shreddedVariant, Schema shreddedSchema, Schema
unshreddedSchema) {
+ GenericRecord rebuilt = new GenericData.Record(unshreddedSchema);
+ rebuilt.put("metadata", shreddedVariant.get("metadata"));
+ rebuilt.put("value", ByteBuffer.wrap(new byte[] {
+ ((Number) shreddedVariant.get("typed_value")).byteValue()}));
+ return rebuilt;
+ }
+ }
}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/parquet/io/TestHoodieParquetBinaryCopyBasePaths.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/parquet/io/TestHoodieParquetBinaryCopyBasePaths.java
new file mode 100644
index 000000000000..8c65ec5dbb39
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/parquet/io/TestHoodieParquetBinaryCopyBasePaths.java
@@ -0,0 +1,76 @@
+/*
+ * 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.parquet.io;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestHoodieParquetBinaryCopyBasePaths {
+
+ @Test
+ void testLegacyArrayPathsAreConverted() {
+ TestableBinaryCopy copy = new TestableBinaryCopy();
+ String[] genericLegacyPath = {"unknown", "bag", "array_element"};
+ assertTrue(copy.convertLegacy3LevelArray(genericLegacyPath));
+ assertArrayEquals(new String[] {"unknown", "list", "element"},
genericLegacyPath);
+
+ copy.requiredSchema = MessageTypeParser.parseMessageType(
+ "message record { optional group values (LIST) { repeated group list {
optional binary element; } } }");
+ String[] avroLegacyPath = {"values", "bag", "array"};
+ assertTrue(copy.convertLegacy3LevelArray(avroLegacyPath));
+ assertArrayEquals(new String[] {"values", "list", "element"},
avroLegacyPath);
+
+ String[] unchanged = {"values", "list", "element"};
+ assertFalse(copy.convertLegacy3LevelArray(unchanged));
+ }
+
+ @Test
+ void testLegacyMapPathsAreConvertedAndUnknownPathsAreIgnored() {
+ TestableBinaryCopy copy = new TestableBinaryCopy();
+ copy.requiredSchema = MessageTypeParser.parseMessageType(
+ "message record { optional group properties (MAP) { repeated group
key_value { "
+ + "required binary key; optional binary value; } } }");
+
+ String[] path = {"properties", "map", "value"};
+ assertTrue(copy.convertLegacyMap(path));
+ assertArrayEquals(new String[] {"properties", "key_value", "value"}, path);
+
+ assertFalse(copy.convertLegacyMap(new String[] {"missing", "map",
"value"}));
+ }
+
+ private static class TestableBinaryCopy extends HoodieParquetBinaryCopyBase {
+ private TestableBinaryCopy() {
+ super(new Configuration());
+ }
+
+ @Override
+ protected Map<String, String> finalizeMetadata() {
+ return Collections.emptyMap();
+ }
+ }
+}
