Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/generated/core_configuration.html
Original file line number Diff line number Diff line change
Expand Up @@ -738,7 +738,7 @@
<td><h5>file.block-size</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>MemorySize</td>
<td>File block size of format, default value of orc stripe is 64 MB, and parquet row group is 128 MB.</td>
<td>File block size of format, default value of orc stripe is 64 MB, parquet row group is 128 MB, and avro block is 64000 bytes. For avro, this is the uncompressed block size threshold.</td>
</tr>
<tr>
<td><h5>file.compression</h5></td>
Expand Down
3 changes: 2 additions & 1 deletion paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
Original file line number Diff line number Diff line change
Expand Up @@ -426,7 +426,8 @@ public InlineElement getDescription() {
.memoryType()
.noDefaultValue()
.withDescription(
"File block size of format, default value of orc stripe is 64 MB, and parquet row group is 128 MB.");
"File block size of format, default value of orc stripe is 64 MB, parquet row group is 128 MB, "
+ "and avro block is 64000 bytes. For avro, this is the uncompressed block size threshold.");

public static final ConfigOption<MemorySize> FILE_INDEX_IN_MANIFEST_THRESHOLD =
key("file-index.in-manifest-threshold")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.ConfigOptions;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.statistics.SimpleColStatsCollector;
Expand Down Expand Up @@ -65,12 +66,14 @@ public class AvroFileFormat extends FileFormat {

private final Options options;
private final int zstdLevel;
@Nullable private final MemorySize blockSize;

public AvroFileFormat(FormatContext context) {
super(IDENTIFIER);

this.options = getIdentifierPrefixOptions(context.options());
this.zstdLevel = context.zstdLevel();
this.blockSize = context.blockSize();
}

@Override
Expand All @@ -93,6 +96,9 @@ public AvroBlockWriter createBlockWriter(
AvroRowDatumWriter datumWriter = new AvroRowDatumWriter(rowType);
DataFileWriter<InternalRow> writer = new DataFileWriter<>(datumWriter);
writer.setCodec(createCodecFactory(compression));
if (blockSize != null) {
writer.setSyncInterval(Math.toIntExact(blockSize.getBytes()));
}
writer.setFlushOnEveryBlock(false);
writer.create(schema, new CloseShieldOutputStream(out));
return new AvroBlockWriter(writer, out, schema);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,9 @@
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;

import java.io.ByteArrayOutputStream;
import java.io.File;
Expand Down Expand Up @@ -76,6 +79,74 @@ public static void before() {
fileFormat = new AvroFileFormat(new FormatContext(new Options(), 1024, 1024));
}

@ParameterizedTest
@CsvSource({
", 64000, null",
", 64000, deflate",
"1 kb, 1024, null",
"1 kb, 1024, deflate",
"4 kb, 4096, null",
"4 kb, 4096, deflate",
"128 kb, 131072, null",
"128 kb, 131072, deflate"
})
void testFileBlockSize(String blockSize, int expectedBlockSize, String compression)
throws IOException {
Options options = new Options();
if (blockSize != null) {
options.setString("file.block-size", blockSize);
}
FileFormat format = FileFormat.fromIdentifier("avro", options);
RowType rowType = DataTypes.ROW(DataTypes.INT().notNull()).notNull();
LocalFileIO fileIO = LocalFileIO.create();
Path file = new Path(new Path(tempPath.toUri()), UUID.randomUUID().toString());
int numRecords = 300_000;

try (PositionOutputStream out = fileIO.newOutputStream(file, false);
FormatWriter writer =
format.createWriterFactory(rowType).create(out, compression)) {
// Each record is encoded as one byte, so record counts also give uncompressed sizes.
for (int i = 0; i < numRecords; i++) {
writer.addElement(GenericRow.of(0));
}
}

long records = 0;
try (AvroBlockReader reader = new AvroBlockReader(fileIO.newInputStream(file))) {
while (reader.hasNextBlock()) {
AvroRawBlock block = reader.nextBorrowedRawBlock();
long expectedRecords = Math.min(expectedBlockSize, numRecords - records);
assertThat(block.recordCount()).isEqualTo(expectedRecords);
assertThat(block.decompress(null).remaining()).isEqualTo((int) expectedRecords);
records += block.recordCount();
}
}
assertThat(records).isEqualTo(numRecords);
}

@ParameterizedTest
@ValueSource(longs = {2147483648L, 4294968320L, Long.MAX_VALUE})
void testFileBlockSizeOverflow(long blockSize) throws IOException {
Options options = new Options();
options.setString("file.block-size", Long.toString(blockSize));
FileFormat format = FileFormat.fromIdentifier("avro", options);
RowType rowType = DataTypes.ROW(DataTypes.INT().notNull()).notNull();
LocalFileIO fileIO = LocalFileIO.create();
Path file = new Path(new Path(tempPath.toUri()), UUID.randomUUID().toString());

try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
assertThatThrownBy(
() -> {
try (FormatWriter writer =
format.createWriterFactory(rowType).create(out, "null")) {
writer.addElement(GenericRow.of(0));
}
})
.isInstanceOf(ArithmeticException.class)
.hasMessage("integer overflow");
}
}

@Test
public void testSupportedDataTypes() {
ArrayList<DataField> dataFields = new ArrayList<>();
Expand Down
Loading