jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[277] added Kryo Integration for Spiller Serialization

This commit is contained in:
Nitin Kashyap 2022-01-27 12:02:55 +05:30
parent 247d7850e6
commit a50d60b2ed
No known key found for this signature in database
GPG Key ID: 7C4222EE7DDEFC35
58 changed files with 1889 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
hetu-docs/en/admin/properties.md
View File

@ -254,6 +254,15 @@ This section describes the most important config properties that may be used to
>
> Sets number of pages prefetched while reading from spilled files.
### `experimental.spill-use-kryo-serialization`
> - **Type:** `boolean`
> - **Default value:** `false`
>
> Enables Kryo based serialization for spill to disk, instead of default java serializer.
### `experimental.revocable-memory-selection-threshold`
> - **Type:** `data size`

4
hetu-hbase/pom.xml
View File

@ -17,7 +17,7 @@
<version.annotations>13.0</version.annotations>
<version.validation-api>2.0.1.Final</version.validation-api>
<version.commons-lang3>3.10</version.commons-lang3>
<version.mockito-all>1.10.18</version.mockito-all>
<version.mockito-all>1.10.19</version.mockito-all>
<version.testing>6.10</version.testing>
<version.hbase>2.2.3</version.hbase>
<version.jacoco-maven-plugin>0.8.2</version.jacoco-maven-plugin>
@ -315,7 +315,7 @@
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<artifactId>mockito-core</artifactId>
<version>${version.mockito-all}</version>
<scope>test</scope>
</dependency>

20
hetu-heuristic-index/pom.xml
View File

@ -16,6 +16,7 @@
<air.main.basedir>${project.parent.basedir}</air.main.basedir>
<air.check.fail-duplicate-finder>false</air.check.fail-duplicate-finder>
<org.slf4j.version>1.7.30</org.slf4j.version>
<powermock.version>2.0.2</powermock.version>
</properties>
<build>
@ -100,36 +101,25 @@
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>3.5.13</version>
<version>1.10.19</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-core</artifactId>
<version>2.0.2</version>
<version>${powermock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-testng</artifactId>
<version>2.0.2</version>
<version>${powermock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-testng-common</artifactId>
<version>2.0.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-api-mockito2</artifactId>
<version>2.0.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.objenesis</groupId>
<artifactId>objenesis</artifactId>
<version>${powermock.version}</version>
<scope>test</scope>
</dependency>
<dependency>

4
hetu-metastore/pom.xml
View File

@ -149,8 +149,8 @@
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<version>1.10.18</version>
<artifactId>mockito-core</artifactId>
<version>1.10.19</version>
<scope>test</scope>
</dependency>
<dependency>

5
hetu-transport/pom.xml
View File

@ -45,5 +45,10 @@
<artifactId>jackson-annotations</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
<version>5.0.3</version>
</dependency>
</dependencies>
</project>

14
hetu-transport/src/main/java/io/hetu/core/transport/block/BlockSerdeUtil.java
View File

@ -13,6 +13,10 @@
*/
package io.hetu.core.transport.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.Slice;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
@ -41,8 +45,18 @@ public final class BlockSerdeUtil
return blockEncodingSerde.readBlock(input);
}
public static Block readBlock(Kryo kryo, Serializer blockEncodingSerde, Input input)
{
return (Block) blockEncodingSerde.read(kryo, input, null);
}
public static void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput output, Block block)
{
blockEncodingSerde.writeBlock(output, block);
}
public static void writeBlock(Kryo kryo, Serializer blockEncodingSerde, Output output, Block block)
{
blockEncodingSerde.write(kryo, output, block);
}
}

113
hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/KryoPageSerializer.java Normal file
View File

@ -0,0 +1,113 @@
/*
* Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.hetu.core.transport.execution.buffer;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.prestosql.spi.Page;
import io.prestosql.spi.block.Block;
import io.prestosql.spi.block.BlockEncodingSerde;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.util.Optional;
import java.util.Properties;
import static com.google.common.base.Preconditions.checkArgument;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
public class KryoPageSerializer
extends PagesSerde
{
BlockEncodingSerde serde;
Serializer<Page> serializer = new Serializer<Page>()
{
@Override
public void write(Kryo kryo, Output output, Page page)
{
output.writeInt(page.getPositionCount());
output.writeInt(page.getChannelCount());
for (int channel = 0; channel < page.getChannelCount(); channel++) {
serde.writeBlock(output, page.getBlock(channel));
}
if (page.getPageMetadata().size() > 0) {
String pageProperties = page.getPageMetadata().toString();
byte[] propertiesByte = pageProperties
.replaceAll(",", System.lineSeparator())
.substring(1, pageProperties.length() - 1)
.getBytes(UTF_8);
output.writeInt(propertiesByte.length);
output.writeBytes(propertiesByte);
}
else {
output.writeInt(0);
}
}
@Override
public Page read(Kryo kryo, Input input, Class<? extends Page> aClass)
{
int positionCount = input.readInt();
int numberOfBlocks = input.readInt();
Block[] blocks = new Block[numberOfBlocks];
for (int i = 0; i < blocks.length; i++) {
blocks[i] = serde.readBlock(input);
}
int propSize = input.readInt();
if (propSize > 0) {
byte[] pageMetadataBytes = input.readBytes(propSize);
Properties pros = new Properties();
try {
pros.load(new ByteArrayInputStream(pageMetadataBytes));
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
return new Page(positionCount, pros, blocks);
}
return new Page(positionCount, blocks);
}
};
KryoPageSerializer(BlockEncodingSerde serde)
{
super(serde, Optional.empty(), Optional.empty(), Optional.empty());
this.serde = requireNonNull(serde, "Serde Cannot be null");
}
@Override
public void serialize(OutputStream output, Page page)
{
checkArgument(output instanceof Output, "Page serializer does not support (" + output.getClass().getSimpleName() + ") for writing");
serializer.write(null, (Output) output, page);
}
@Override
public Page deserialize(InputStream input)
{
checkArgument(input instanceof Input, "Page serializer does not support (" + input.getClass().getSimpleName() + ") for reading");
return serializer.read(null, (Input) input, Page.class);
}
}

12
hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeFactory.java
View File

@ -35,17 +35,21 @@ public class PagesSerdeFactory
public PagesSerde createPagesSerde()
{
return createPagesSerdeInternal(Optional.empty(), false);
return createPagesSerdeInternal(Optional.empty(), false, false);
}
public PagesSerde createPagesSerdeForSpill(Optional<SpillCipher> spillCipher, boolean useDirectSerde)
public PagesSerde createPagesSerdeForSpill(Optional<SpillCipher> spillCipher, boolean useDirectSerde, boolean useKryo)
{
return createPagesSerdeInternal(spillCipher, useDirectSerde);
return createPagesSerdeInternal(spillCipher, useDirectSerde, useKryo);
}
private PagesSerde createPagesSerdeInternal(Optional<SpillCipher> spillCipher, boolean useDirectSerde)
private PagesSerde createPagesSerdeInternal(Optional<SpillCipher> spillCipher, boolean useDirectSerde, boolean useKryo)
{
if (useDirectSerde) {
if (useKryo) {
return new KryoPageSerializer(blockEncodingSerde);
}
return new SliceStreamPageSerde(blockEncodingSerde, Optional.empty(), Optional.empty(), spillCipher);
}

23
hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeUtil.java
View File

@ -13,6 +13,10 @@
*/
package io.hetu.core.transport.execution.buffer;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.common.collect.AbstractIterator;
import io.airlift.slice.Slice;
import io.airlift.slice.SliceInput;
@ -46,6 +50,25 @@ public class PagesSerdeUtil
{
}
static void writeRawPage(Kryo kryo, Page page, Output output, Serializer serde)
{
output.write(page.getChannelCount());
for (int channel = 0; channel < page.getChannelCount(); channel++) {
writeBlock(kryo, serde, output, page.getBlock(channel));
}
}
static Page readRawPage(Kryo kryo, int positionCount, Input input, Serializer blockEncodingSerde)
{
int numberOfBlocks = input.readInt();
Block[] blocks = new Block[numberOfBlocks];
for (int i = 0; i < blocks.length; i++) {
blocks[i] = readBlock(kryo, blockEncodingSerde, input);
}
return new Page(positionCount, blocks);
}
static void writeRawPage(Page page, SliceOutput output, BlockEncodingSerde serde)
{
output.writeInt(page.getChannelCount());

2
pom.xml
View File

@ -1491,7 +1491,7 @@
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>1.9.5</version>
<version>1.10.19</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>

27
presto-main/pom.xml
View File

@ -305,8 +305,21 @@
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
<version>1.10.19</version>
<exclusions>
<exclusion>
<groupId>org.objenesis</groupId>
<artifactId>objenesis</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.hamcrest</groupId>
<artifactId>hamcrest-core</artifactId>
<version>1.1</version>
<scope>test</scope>
</dependency>
@ -423,6 +436,12 @@
<artifactId>bcprov-jdk15on</artifactId>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
<version>5.0.3</version>
</dependency>
<!-- for testing -->
<dependency>
<groupId>org.testng</groupId>
@ -466,6 +485,12 @@
<artifactId>presto-spi</artifactId>
<type>test-jar</type>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>

13
presto-main/src/main/java/io/prestosql/metadata/FunctionAndTypeManager.java
View File

@ -13,6 +13,7 @@
*/
package io.prestosql.metadata;
import com.esotericsoftware.kryo.Kryo;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
@ -126,13 +127,15 @@ public class FunctionAndTypeManager
private final LoadingCache<FunctionResolutionCacheKey, FunctionHandle> functionCache;
private final CacheStatsMBean cacheStatsMBean;
private final ConcurrentMap<String, BlockEncoding> blockEncodings = new ConcurrentHashMap<>();
private final Kryo kryo;
@Inject
public FunctionAndTypeManager(
TransactionManager transactionManager,
FeaturesConfig featuresConfig,
HandleResolver handleResolver,
Set<Type> types)
Set<Type> types,
Kryo kryo)
{
this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
this.builtInFunctionNamespaceManager = new BuiltInFunctionNamespaceManager(featuresConfig, this);
@ -148,6 +151,7 @@ public class FunctionAndTypeManager
.build(CacheLoader.from(key -> resolveBuiltInFunction(key.functionName, fromTypeSignatures(key.parameterTypes))));
this.cacheStatsMBean = new CacheStatsMBean(functionCache);
this.functionResolver = new FunctionResolver(this);
this.kryo = requireNonNull(kryo, "Kryo Object cannot be null");
// add the built-in BlockEncodings
addBlockEncoding(new VariableWidthBlockEncoding());
@ -168,7 +172,7 @@ public class FunctionAndTypeManager
public static FunctionAndTypeManager createTestFunctionAndTypeManager()
{
return new FunctionAndTypeManager(createTestTransactionManager(), new FeaturesConfig(), new HandleResolver(), ImmutableSet.of());
return new FunctionAndTypeManager(createTestTransactionManager(), new FeaturesConfig(), new HandleResolver(), ImmutableSet.of(), new Kryo());
}
public BlockEncoding getBlockEncoding(String encodingName)
@ -183,6 +187,11 @@ public class FunctionAndTypeManager
return new InternalBlockEncodingSerde(this);
}
public BlockEncodingSerde getBlockKryoEncodingSerde()
{
return new KryoBlockEncodingSerde(this, kryo);
}
public void addBlockEncoding(BlockEncoding blockEncoding)
{
requireNonNull(blockEncoding, "blockEncoding is null");

2
presto-main/src/main/java/io/prestosql/metadata/InternalBlockEncodingSerde.java
View File

@ -75,7 +75,7 @@ final class InternalBlockEncodingSerde
}
}
private static String readLengthPrefixedString(SliceInput input)
protected static String readLengthPrefixedString(SliceInput input)
{
int length = input.readInt();
byte[] bytes = new byte[length];

137
presto-main/src/main/java/io/prestosql/metadata/KryoBlockEncodingSerde.java Normal file
View File

@ -0,0 +1,137 @@
/*
* Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.metadata;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import io.prestosql.spi.block.AbstractBlockEncoding;
import io.prestosql.spi.block.Block;
import io.prestosql.spi.block.BlockEncoding;
import io.prestosql.spi.block.BlockEncodingSerde;
import java.io.InputStream;
import java.io.OutputStream;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
public class KryoBlockEncodingSerde
implements BlockEncodingSerde
{
private final FunctionAndTypeManager functionAndTypeManager;
private final Kryo kryo;
public KryoBlockEncodingSerde(FunctionAndTypeManager metadata, Kryo kryo)
{
this.functionAndTypeManager = requireNonNull(metadata, "metadata is null");
this.kryo = kryo; /*Todo(nitin) make it singleton inject time initializer */
}
public Kryo getKryo()
{
return kryo;
}
@Override
public Object getContext()
{
return getKryo();
}
/**
* Read a block encoding from the input.
*
* @param inputStream
*/
@Override
public Block readBlock(InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR,
"This interface should not be called in this flow");
}
Input input = (Input) inputStream;
String encodingName;
// read the encoding name
encodingName = readLengthPrefixedString((Input) input);
// look up the encoding factory
BlockEncoding blockEncoding = functionAndTypeManager.getBlockEncoding(encodingName);
Serializer<?> serializer = getSerializerFromBlockEncoding(blockEncoding);
if (serializer == null) {
return (Block) blockEncoding.readBlock(this, inputStream);
}
return (Block) serializer.read(kryo, input, null);
}
/**
* Write a blockEncoding to the output.
*
* @param outputStream
* @param block
*/
@Override
public void writeBlock(OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR,
"This interface should not be called in this flow");
}
Output output = (Output) outputStream;
String encodingName = block.getEncodingName();
BlockEncoding blockEncoding = functionAndTypeManager.getBlockEncoding(encodingName);
Serializer<Block<?>> serializer = getSerializerFromBlockEncoding(blockEncoding);
// write the name to the output
writeLengthPrefixedString(output, encodingName);
if (serializer == null) {
blockEncoding.writeBlock(this, outputStream, block);
}
else {
// write the block to the output
serializer.write(kryo, output, block);
}
}
private Serializer<Block<?>> getSerializerFromBlockEncoding(BlockEncoding blockEncoding)
{
if (blockEncoding instanceof AbstractBlockEncoding) {
return (Serializer<Block<?>>) blockEncoding;
}
return null;
}
private static String readLengthPrefixedString(Input input)
{
int length = input.readInt();
byte[] bytes = input.readBytes(length);
return new String(bytes, UTF_8);
}
private static void writeLengthPrefixedString(Output output, String value)
{
byte[] bytes = value.getBytes(UTF_8);
output.writeInt(bytes.length);
output.writeBytes(bytes);
}
}

3
presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java
View File

@ -13,6 +13,7 @@
*/
package io.prestosql.metadata;
import com.esotericsoftware.kryo.Kryo;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.collect.HashMultimap;
@ -201,7 +202,7 @@ public final class MetadataManager
public static MetadataManager createTestMetadataManager(TransactionManager transactionManager, FeaturesConfig featuresConfig)
{
return new MetadataManager(
new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of()),
new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of(), new Kryo()),
featuresConfig,
new SessionPropertyManager(),
new SchemaPropertyManager(),

3
presto-main/src/main/java/io/prestosql/server/ServerMainModule.java
View File

@ -13,6 +13,7 @@
*/
package io.prestosql.server;
import com.esotericsoftware.kryo.Kryo;
import com.google.common.collect.ImmutableList;
import com.google.inject.Binder;
import com.google.inject.Key;
@ -439,6 +440,8 @@ public class ServerMainModule
jsonBinder(binder).addDeserializerBinding(Type.class).to(TypeDeserializer.class);
newSetBinder(binder, Type.class);
binder.bind(Kryo.class).in(Scopes.SINGLETON);
// split manager
binder.bind(SplitManager.class).in(Scopes.SINGLETON);

24
presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpiller.java
View File

@ -13,6 +13,8 @@
*/
package io.prestosql.spiller;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Stopwatch;
import com.google.common.collect.AbstractIterator;
@ -89,6 +91,7 @@ public class FileSingleStreamSpiller
private long spilledPagesInMemorySize;
private ListenableFuture<?> spillInProgress = Futures.immediateFuture(null);
private boolean useDirectSerde;
private boolean useKryo;
private boolean compressionEnabled;
private Optional<SpillCipher> spillCipher;
private byte[] cipherIV;
@ -107,7 +110,8 @@ public class FileSingleStreamSpiller
Optional<SpillCipher> spillCipher,
boolean compressionEnabled,
boolean useDirectSerde,
int spillPrefetchReadPages)
int spillPrefetchReadPages,
boolean useKryo)
{
this.serde = requireNonNull(serde, "serde is null");
this.executor = requireNonNull(executor, "executor is null");
@ -136,6 +140,7 @@ public class FileSingleStreamSpiller
}
this.spillPrefetchReadPages = spillPrefetchReadPages;
this.useDirectSerde = useDirectSerde;
this.useKryo = useKryo;
this.compressionEnabled = compressionEnabled;
this.spillCipher = spillCipher;
}
@ -206,9 +211,14 @@ public class FileSingleStreamSpiller
System.arraycopy(tmpCipherIV, 0, this.cipherIV, 0, tmpCipherIV.length);
tmpOutputStream = new CipherOutputStream(tmpOutputStream, cipher);
}
if (compressionEnabled) {
tmpOutputStream = new SnappyFramedOutputStream(tmpOutputStream);
}
if (useKryo) {
return new Output(tmpOutputStream, bufferSize);
}
return new OutputStreamSliceOutput(tmpOutputStream, bufferSize);
}
@ -240,14 +250,23 @@ public class FileSingleStreamSpiller
if (spillCipher.isPresent()) {
tmpInputStream = new CipherInputStream(tmpInputStream, spillCipher.get().getDecryptionCipher(cipherIV));
}
if (compressionEnabled) {
tmpInputStream = new SnappyFramedInputStream(tmpInputStream);
}
if (useKryo) {
return new Input(tmpInputStream, bufferSize);
}
return new InputStreamSliceInput(tmpInputStream, bufferSize);
}
private Predicate<InputStream> getStreamEndOfData()
{
if (useKryo) {
return (input) -> ((Input) input).end();
}
return (input) -> !((InputStreamSliceInput) input).isReadable();
}
@ -348,6 +367,7 @@ public class FileSingleStreamSpiller
state.targetFile = this.targetFile.getFilePath().toAbsolutePath().toString();
state.compressionEnabled = this.compressionEnabled;
state.useDirectSerde = this.useDirectSerde;
state.useKryo = this.useKryo;
return state;
}
@ -371,6 +391,7 @@ public class FileSingleStreamSpiller
}
this.compressionEnabled = myState.compressionEnabled;
this.useDirectSerde = myState.useDirectSerde;
this.useKryo = myState.useKryo;
}
catch (IOException e) {
throw new UncheckedIOException(e);
@ -386,5 +407,6 @@ public class FileSingleStreamSpiller
private String targetFile;
private boolean compressionEnabled;
private boolean useDirectSerde;
private boolean useKryo;
}
}

35
presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpillerFactory.java
View File

@ -21,6 +21,7 @@ import io.airlift.log.Logger;
import io.hetu.core.transport.execution.buffer.PagesSerde;
import io.hetu.core.transport.execution.buffer.PagesSerdeFactory;
import io.prestosql.memory.context.LocalMemoryContext;
import io.prestosql.metadata.KryoBlockEncodingSerde;
import io.prestosql.metadata.Metadata;
import io.prestosql.operator.SpillContext;
import io.prestosql.spi.PrestoException;
@ -39,6 +40,7 @@ import java.nio.file.Path;
import java.util.List;
import java.util.Optional;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.util.concurrent.MoreExecutors.listeningDecorator;
import static io.airlift.concurrent.Threads.daemonThreadsNamed;
import static io.prestosql.spi.StandardErrorCode.OUT_OF_SPILL_SPACE;
@ -68,6 +70,7 @@ public class FileSingleStreamSpillerFactory
private final double maxUsedSpaceThreshold;
private final boolean spillEncryptionEnabled;
private final boolean spillDirectSerdeEnabled;
private final boolean useKryo;
private final boolean spillCompressionEnabled;
private int roundRobinIndex;
private int spillPrefetchReadPages;
@ -79,14 +82,15 @@ public class FileSingleStreamSpillerFactory
listeningDecorator(newFixedThreadPool(
requireNonNull(featuresConfig, "featuresConfig is null").getSpillerThreads(),
daemonThreadsNamed("binary-spiller-%s"))),
requireNonNull(metadata, "metadata is null").getFunctionAndTypeManager().getBlockEncodingSerde(),
requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillUseKryoSerialization() ? requireNonNull(metadata, "metadata is null").getFunctionAndTypeManager().getBlockKryoEncodingSerde() : requireNonNull(metadata, "metadata is null").getFunctionAndTypeManager().getBlockEncodingSerde(),
spillerStats,
requireNonNull(featuresConfig, "featuresConfig is null").getSpillerSpillPaths(),
requireNonNull(featuresConfig, "featuresConfig is null").getSpillMaxUsedSpaceThreshold(),
requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillCompressionEnabled(),
requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillEncryptionEnabled(),
requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillDirectSerdeEnabled(),
requireNonNull(nodeSpillConfig, "featuresConfig is null").getSpillPrefetchReadPages());
requireNonNull(nodeSpillConfig, "featuresConfig is null").getSpillPrefetchReadPages(),
requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillUseKryoSerialization());
}
@VisibleForTesting
@ -101,6 +105,28 @@ public class FileSingleStreamSpillerFactory
boolean spillDirectSerdeEnabled,
int spillPrefetchReadPages)
{
this(executor, blockEncodingSerde, spillerStats, spillPaths, maxUsedSpaceThreshold,
spillCompressionEnabled, spillEncryptionEnabled, spillDirectSerdeEnabled,
spillPrefetchReadPages, false);
}
@VisibleForTesting
public FileSingleStreamSpillerFactory(
ListeningExecutorService executor,
BlockEncodingSerde blockEncodingSerde,
SpillerStats spillerStats,
List<Path> spillPaths,
double maxUsedSpaceThreshold,
boolean spillCompressionEnabled,
boolean spillEncryptionEnabled,
boolean spillDirectSerdeEnabled,
int spillPrefetchReadPages,
boolean useKryo)
{
checkArgument(!(blockEncodingSerde instanceof KryoBlockEncodingSerde)
|| (blockEncodingSerde instanceof KryoBlockEncodingSerde && spillDirectSerdeEnabled),
"Kryo serialization should enable DirectSpill");
this.serdeFactory = new PagesSerdeFactory(blockEncodingSerde, spillCompressionEnabled);
this.executor = requireNonNull(executor, "executor is null");
this.spillerStats = requireNonNull(spillerStats, "spillerStats can not be null");
@ -125,6 +151,7 @@ public class FileSingleStreamSpillerFactory
this.roundRobinIndex = 0;
this.spillDirectSerdeEnabled = spillDirectSerdeEnabled;
this.spillPrefetchReadPages = spillPrefetchReadPages;
this.useKryo = useKryo;
}
@PostConstruct
@ -164,8 +191,8 @@ public class FileSingleStreamSpillerFactory
if (spillEncryptionEnabled) {
spillCipher = Optional.of(new AesSpillCipher());
}
PagesSerde serde = serdeFactory.createPagesSerdeForSpill(spillCipher, spillDirectSerdeEnabled);
return new FileSingleStreamSpiller(serde, executor, getNextSpillPath(), spillerStats, spillContext, memoryContext, spillCipher, spillCompressionEnabled, spillDirectSerdeEnabled, spillPrefetchReadPages);
PagesSerde serde = serdeFactory.createPagesSerdeForSpill(spillCipher, spillDirectSerdeEnabled, useKryo);
return new FileSingleStreamSpiller(serde, executor, getNextSpillPath(), spillerStats, spillContext, memoryContext, spillCipher, spillCompressionEnabled, spillDirectSerdeEnabled, spillPrefetchReadPages, useKryo);
}
private synchronized Path getNextSpillPath()

13
presto-main/src/main/java/io/prestosql/spiller/NodeSpillConfig.java
View File

@ -31,6 +31,7 @@ public class NodeSpillConfig
private boolean spillDirectSerdeEnabled;
private int spillPrefetchReadPages = 1;
private boolean spillUseKryoSerialization;
@NotNull
public DataSize getMaxSpillPerNode()
@ -107,4 +108,16 @@ public class NodeSpillConfig
this.spillPrefetchReadPages = spillPrefetchedReadPages;
return this;
}
public boolean isSpillUseKryoSerialization()
{
return spillUseKryoSerialization;
}
@Config("experimental.spill-use-kryo-serialization")
public NodeSpillConfig setSpillUseKryoSerialization(boolean spillUseKryoSerialization)
{
this.spillUseKryoSerialization = spillUseKryoSerialization;
return this;
}
}

3
presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java
View File

@ -13,6 +13,7 @@
*/
package io.prestosql.testing;
import com.esotericsoftware.kryo.Kryo;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
@ -340,7 +341,7 @@ public class LocalQueryRunner
this.planOptimizerManager = new ConnectorPlanOptimizerManager();
this.metadata = new MetadataManager(
new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of()),
new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of(), new Kryo()),
featuresConfig,
// new HetuConfig object passed, if split filtering is needed in the runner, a modified HetuConfig object with filter settings manually set must be used.
new SessionPropertyManager(new SystemSessionProperties(new QueryManagerConfig(), taskManagerConfig, new MemoryManagerConfig(), featuresConfig, new HetuConfig(), new SnapshotConfig())),

2
presto-main/src/test/java/io/prestosql/operator/TestGroupByHash.java
View File

@ -331,7 +331,7 @@ public class TestGroupByHash
expectedMapping.put("hashCollisions", 37L);
expectedMapping.put("expectedHashCollisions", 0.0);
expectedMapping.put("preallocatedMemoryInBytes", 0L);
expectedMapping.put("currentPageSizeInBytes", 4732L);
expectedMapping.put("currentPageSizeInBytes", 4740L);
return expectedMapping;
}

4
presto-main/src/test/java/io/prestosql/operator/TestHashAggregationOperator.java
View File

@ -317,7 +317,7 @@ public class TestHashAggregationOperator
//TODO-cp-I2DSGQ: change expectedMapping after implementation of operatorContext capture
expectedMapping.put("operatorContext", 0);
expectedMapping.put("aggregationBuilder", aggregationBuilderMapping);
expectedMapping.put("memoryContext", 10355419L);
expectedMapping.put("memoryContext", 10675419L);
expectedMapping.put("inputProcessed", true);
expectedMapping.put("finishing", false);
expectedMapping.put("finished", false);
@ -405,7 +405,7 @@ public class TestHashAggregationOperator
aggregation3Array0.put("capacity", 40960);
aggregation3Array0.put("segments", 40);
aggregation3Array1.put("array", blockObjectBigArrayState);
aggregation3Array1.put("sizeOfBlocks", 2504108L);
aggregation3Array1.put("sizeOfBlocks", 2824108L);
blockObjectBigArrayState.put("array", 1);
blockObjectBigArrayState.put("capacity", 40960);
aggregation3StateList.add(39999L);

2
presto-main/src/test/java/io/prestosql/operator/TestOrderByOperator.java
View File

@ -266,7 +266,7 @@ public class TestOrderByOperator
Map<String, Object> expectedMapping = new HashMap<>();
expectedMapping.put("operatorContext", 0);
expectedMapping.put("revocableMemoryContext", 0L);
expectedMapping.put("localUserMemoryContext", 8828L);
expectedMapping.put("localUserMemoryContext", 8844L);
return expectedMapping;
}

4
presto-main/src/test/java/io/prestosql/operator/TestPagesIndex.java
View File

@ -80,8 +80,8 @@ public class TestPagesIndex
expectedMapping.put("valueAddresses", valueAddresses);
expectedMapping.put("nextBlockToCompact", 0);
expectedMapping.put("positionCount", 7);
expectedMapping.put("pagesMemorySize", 3852L);
expectedMapping.put("estimatedSize", 12396L);
expectedMapping.put("pagesMemorySize", 3860L);
expectedMapping.put("estimatedSize", 12404L);
return expectedMapping;
}

2
presto-main/src/test/java/io/prestosql/operator/TestStreamingAggregationOperator.java
View File

@ -178,7 +178,7 @@ public class TestStreamingAggregationOperator
Map<String, Object> expectedMapping = new HashMap<>();
expectedMapping.put("operatorContext", 0);
expectedMapping.put("systemMemoryContext", 0L);
expectedMapping.put("userMemoryContext", 2236L);
expectedMapping.put("userMemoryContext", 2244L);
expectedMapping.put("finishing", false);
return expectedMapping;
}

4
presto-main/src/test/java/io/prestosql/operator/TestTopNOperator.java
View File

@ -179,7 +179,7 @@ public class TestTopNOperator
expectedMapping.put("operatorContext", 0);
expectedMapping.put("workProcessorOperator", workProcessorOperatorMapping);
workProcessorOperatorMapping.put("localUserMemoryContext", 19152L);
workProcessorOperatorMapping.put("localUserMemoryContext", 19168L);
workProcessorOperatorMapping.put("topNBuilder", topNBuilderMapping);
workProcessorOperatorMapping.put("outputIterator", false);
@ -187,7 +187,7 @@ public class TestTopNOperator
topNBuilderMapping.put("groupedRows", groupedRowsMapping);
topNBuilderMapping.put("pageReferences", pageReferencesMapping);
topNBuilderMapping.put("emptyPageReferenceSlots", emptyPageReferenceSlots);
topNBuilderMapping.put("memorySizeInBytes", 2496L);
topNBuilderMapping.put("memorySizeInBytes", 2512L);
topNBuilderMapping.put("currentPageCount", 2);
groupedRowsMapping.put("array", Object[][].class);

112
presto-main/src/test/java/io/prestosql/spiller/TestFileSingleStreamSpiller.java
View File

@ -186,29 +186,55 @@ public class TestFileSingleStreamSpiller
public void testSpillWithSingleFile()
throws Exception
{
assertSpillBenchmark(false, false, "1GB", 1, false);
assertSpillBenchmark(false, false, "1GB", 1, false, false);
}
@Test
public void testSpillWithMultiFile()
throws Exception
{
assertSpillBenchmark(false, false, "2MB", 500, false);
assertSpillBenchmark(false, false, "2MB", 500, false, false);
}
private void assertSpillBenchmark(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useDirectSerde)
@Test
public void testSpillWithSingleFileWithKryo()
throws Exception
{
assertSpillBenchmark(false, false, "1GB", 1, true, true);
}
@Test
public void testSpillWithMultiFileWithKryo()
throws Exception
{
assertSpillBenchmark(false, false, "2MB", 2, true, true);
}
@Test
public void testSpillWithSingleSpillerConsolidatedWithoutWorkProcessor()
throws Exception
{
assertSpillBenchmark(false, false, "1GB", 1, false, false);
assertSpillBenchmark(false, false, "1GB", 1, true, true);
assertSpillBenchmark(false, false, "2MB", 512, false, false);
assertSpillBenchmark(false, false, "2MB", 512, true, true);
}
private void assertSpillBenchmark(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useKryo, boolean useDirectSerde)
throws Exception
{
List<FileSingleStreamSpiller> spillers = new ArrayList<>();
FileSingleStreamSpillerFactory spillerFactory = new FileSingleStreamSpillerFactory(
executorBenchmark, // executor won't be closed, because we don't call destroy() on the spiller factory
createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(),
(useKryo) ? createTestMetadataManager().getFunctionAndTypeManager().getBlockKryoEncodingSerde() : createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(),
new SpillerStats(),
ImmutableList.of(spillPath.toPath()),
1.0,
compression,
encryption,
useDirectSerde, 1);
useDirectSerde,
1,
useKryo);
LocalMemoryContext memoryContext = newSimpleAggregatedMemoryContext().newLocalMemoryContext("test");
long startTime = System.currentTimeMillis();
Stopwatch spillTimer = Stopwatch.createStarted();
@ -285,81 +311,81 @@ public class TestFileSingleStreamSpiller
return new Page(col1.build(), col2.build());
}
@Test
public void testSpillWithSingleSpillerConsolidated()
throws Exception
{
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 1, false);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 1, true);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 1, false);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 1, true);
}
@Test
public void testSpillWithSingleSpillerConsolidatedWithCompression()
throws Exception
{
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, false, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, false, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, false, 1, false);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 1, false);
}
@Test
public void testSpillWithSingleSpillerConsolidatedWithCompressionMultiFile()
throws Exception
{
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, false, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, true, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, false, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, true, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, false, 1, false);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, true, 1, false);
}
@Test
public void testSpillWithSingleSpillerConsolidatedWithoutCompression()
throws Exception
{
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 25, false);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 25, false);
}
@Test
public void testSpillWithSingleSpillerConsolidatedWithoutCompressionMultiFile()
throws Exception
{
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 25, false);
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 25, false);
}
@Test
public void testSpillWithSingleSpillerConsolidatedWithEncryption()
throws Exception
{
assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, false, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, true, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, false, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 1);
assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, false, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, true, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, false, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 25);
assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, false, 25, false);
assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 25, false);
}
private void assertSpillBenchmarkReadingUsingWorkProcessor(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useDirectSerde, int spillPrefetchReadPages)
@Test
public void testSpillWithSingleSpillerConsolidatedDirectWriteCompareWithWorkProcessorWithKryo()
throws Exception
{
assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 1, true);
assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 1, true);
assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, true, 1, true);
assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 1, true);
}
private void assertSpillBenchmarkReadingUsingWorkProcessor(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useDirectSerde, int spillPrefetchReadPages, boolean useKryo)
throws Exception
{
List<FileSingleStreamSpiller> spillers = new ArrayList<>();
FileSingleStreamSpillerFactory spillerFactory = new FileSingleStreamSpillerFactory(
executorBenchmark, // executor won't be closed, because we don't call destroy() on the spiller factory
createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(),
(useKryo) ? createTestMetadataManager().getFunctionAndTypeManager().getBlockKryoEncodingSerde() : createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(),
new SpillerStats(),
ImmutableList.of(spillPath.toPath()),
1.0,
compression,
encryption,
useDirectSerde,
spillPrefetchReadPages);
spillPrefetchReadPages,
useKryo);
LocalMemoryContext memoryContext = newSimpleAggregatedMemoryContext().newLocalMemoryContext("test");
long startTime = System.currentTimeMillis();
@ -420,13 +446,9 @@ public class TestFileSingleStreamSpiller
}
spillTimer.stop();
long timeToUnspill = spillTimer.elapsed(TimeUnit.MILLISECONDS);
/*
* System.out.println(String.format("[isEncrypted: %6s, isCompressed: %6s, isDirect: %6s, SpillSize: %4s, SpillFiles: %4d] --> Spill: %8d, Unspill: %8d",
* encryption, compression, useDirectSerde, pageSize, fileCount, timeToSpill, timeToUnspill, humanReadableByteCountBin(spilledDiskSize)));
*/
log.debug("TimeTakenReadingFromSpill = " + (System.currentTimeMillis() - startTime));
log.info("[isEncrypted: %6s, isCompressed: %6s, isDirect: %6s, SpillSize: %4s, SpillFiles: %4d] --> Spill: %8d, Unspill: %8d, DiskUsed: %5s",
encryption, compression, useDirectSerde, pageSize, fileCount,
log.info("[isEncrypted: %6s, isCompressed: %6s, isDirect: %6s, SpillSize: %4s, SpillFiles: %4d, useKryo: %6s] --> Spill: %8d, Unspill: %8d, DiskUsed: %5s",
encryption, compression, useDirectSerde, pageSize, fileCount, useKryo,
timeToSpill, timeToUnspill, humanReadableByteCountBin(spilledDiskSize));
spillers.stream().forEach(spiller -> spiller.close());
assertEquals(listFiles(spillPath.toPath()).size(), 0);

7
presto-main/src/test/java/io/prestosql/spiller/TestNodeSpillConfig.java
View File

@ -36,7 +36,8 @@ public class TestNodeSpillConfig
.setSpillCompressionEnabled(false)
.setSpillEncryptionEnabled(false)
.setSpillDirectSerdeEnabled(false)
.setSpillPrefetchReadPages(1));
.setSpillPrefetchReadPages(1)
.setSpillUseKryoSerialization(false));
}
@Test
@ -49,6 +50,7 @@ public class TestNodeSpillConfig
.put("experimental.spill-encryption-enabled", "true")
.put("experimental.spill-direct-serde-enabled", "true")
.put("experimental.spill-prefetch-read-pages", "25")
.put("experimental.spill-use-kryo-serialization", "true")
.build();
NodeSpillConfig expected = new NodeSpillConfig()
@ -57,7 +59,8 @@ public class TestNodeSpillConfig
.setSpillCompressionEnabled(true)
.setSpillEncryptionEnabled(true)
.setSpillDirectSerdeEnabled(true)
.setSpillPrefetchReadPages(25);
.setSpillPrefetchReadPages(25)
.setSpillUseKryoSerialization(true);
assertFullMapping(properties, expected);
}

2
presto-main/src/test/java/io/prestosql/spiller/TestSpillCipherPagesSerde.java
View File

@ -41,7 +41,7 @@ public class TestSpillCipherPagesSerde
public void test()
{
SpillCipher cipher = new AesSpillCipher();
PagesSerde serde = TESTING_SERDE_FACTORY.createPagesSerdeForSpill(Optional.of(cipher), false);
PagesSerde serde = TESTING_SERDE_FACTORY.createPagesSerdeForSpill(Optional.of(cipher), false, false);
List<Type> types = ImmutableList.of(VARCHAR);
Page emptyPage = new Page(VARCHAR.createBlockBuilder(null, 0).build());
assertPageEquals(types, serde.deserialize(serde.serialize(emptyPage)), emptyPage);

5
presto-spi/pom.xml
View File

@ -145,5 +145,10 @@
<groupId>io.airlift</groupId>
<artifactId>joni</artifactId>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
<version>5.0.3</version>
</dependency>
</dependencies>
</project>

23
presto-spi/src/main/java/io/prestosql/spi/block/AbstractBlockEncoding.java Normal file
View File

@ -0,0 +1,23 @@
/*
* Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.spi.block;
import com.esotericsoftware.kryo.Serializer;
public abstract class AbstractBlockEncoding<T>
extends Serializer<T>
implements BlockEncoding
{
}

58
presto-spi/src/main/java/io/prestosql/spi/block/ArrayBlockEncoding.java
View File

@ -13,9 +13,16 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.airlift.slice.Slices;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.prestosql.spi.block.ArrayBlock.createArrayBlockInternal;
import static io.prestosql.spi.block.EncoderUtil.decodeNullBits;
@ -65,4 +72,55 @@ public class ArrayBlockEncoding
boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElseGet(() -> new boolean[positionCount]);
return createArrayBlockInternal(0, positionCount, valueIsNull, offsets, values);
}
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleMap ReadBlock");
}
Input input = (Input) inputStream;
Block values = blockEncodingSerde.readBlock(input);
int positionCount = input.readInt();
int[] offsets = input.readInts(positionCount);
boolean[] valueIsNull = null;
if (input.readBoolean()) {
valueIsNull = input.readBooleans(positionCount);
}
return createArrayBlockInternal(0, positionCount, valueIsNull, offsets, values);
}
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock");
}
Output output = (Output) outputStream;
AbstractArrayBlock arrayBlock = (AbstractArrayBlock) block;
int positionCount = arrayBlock.getPositionCount();
int offsetBase = arrayBlock.getOffsetBase();
int[] offsets = arrayBlock.getOffsets();
int valuesStartOffset = offsets[offsetBase];
int valuesEndOffset = offsets[offsetBase + positionCount];
Block values = arrayBlock.getRawElementBlock().getRegion(valuesStartOffset, valuesEndOffset - valuesStartOffset);
blockEncodingSerde.writeBlock(output, values);
output.writeInt(positionCount);
for (int position = 0; position < positionCount + 1; position++) {
output.writeInt(offsets[offsetBase + position] - valuesStartOffset);
}
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(arrayBlock.getValueIsNull(), offsetBase, positionCount);
}
}
}

13
presto-spi/src/main/java/io/prestosql/spi/block/BlockEncoding.java
View File

@ -16,6 +16,8 @@ package io.prestosql.spi.block;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Optional;
public interface BlockEncoding
@ -36,6 +38,17 @@ public interface BlockEncoding
*/
void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceOutput, Block block);
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*/
Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input);
/**
* Write the specified block to the specified output
*/
void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block);
/**
* This method allows the implementor to specify a replacement object that will be serialized instead of the original one.
*/

37
presto-spi/src/main/java/io/prestosql/spi/block/BlockEncodingSerde.java
View File

@ -15,16 +15,49 @@ package io.prestosql.spi.block;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
public interface BlockEncodingSerde
{
/**
* Read a block encoding from the input.
*/
Block readBlock(SliceInput input);
default Block readBlock(InputStream input)
{
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported");
}
/**
* Write a blockEncoding to the output.
*/
void writeBlock(SliceOutput output, Block block);
default void writeBlock(OutputStream output, Block block)
{
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported");
}
/**
* Read a block encoding from the input.
*/
default Block readBlock(SliceInput input)
{
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported");
}
/**
* Write a blockEncoding to the output.
*/
default void writeBlock(SliceOutput output, Block block)
{
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported");
}
/* give out context object if any */
default Object getContext()
{
return null;
}
}

6
presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlock.java
View File

@ -35,11 +35,11 @@ public class ByteArrayBlock
{
private static final int INSTANCE_SIZE = ClassLayout.parseClass(ByteArrayBlock.class).instanceSize();
private final int arrayOffset;
protected final int arrayOffset;
private final int positionCount;
@Nullable
private final boolean[] valueIsNull;
private final byte[] values;
protected final boolean[] valueIsNull;
protected final byte[] values;
private final long sizeInBytes;
private final long retainedSizeInBytes;

69
presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlockEncoding.java
View File

@ -13,14 +13,22 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.prestosql.spi.block.EncoderUtil.decodeNullBits;
import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits;
public class ByteArrayBlockEncoding
implements BlockEncoding
extends AbstractBlockEncoding<ByteArrayBlock>
{
public static final String NAME = "BYTE_ARRAY";
@ -61,4 +69,63 @@ public class ByteArrayBlockEncoding
return new ByteArrayBlock(0, positionCount, valueIsNull, values);
}
@Override
public void write(Kryo kryo, Output output, ByteArrayBlock block)
{
int positionCount = block.getPositionCount();
output.writeInt(positionCount);
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(block.valueIsNull, 0, positionCount);
}
output.writeBytes(block.values, block.arrayOffset, positionCount);
}
@Override
public ByteArrayBlock read(Kryo kryo, Input input, Class<? extends ByteArrayBlock> aClass)
{
int positionCount = input.readInt();
boolean[] valueIsNull = null;
if (input.readBoolean()) {
valueIsNull = input.readBooleans(positionCount);
}
byte[] values = input.readBytes(positionCount);
return new ByteArrayBlock(0, positionCount, valueIsNull, values);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param input
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for ByteArrayBlock");
}
return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, ByteArrayBlock.class);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param output
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for ByteArrayBlock");
}
this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (ByteArrayBlock) block);
}
}

4
presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlock.java
View File

@ -41,8 +41,8 @@ public class DictionaryBlock<T>
private final int positionCount;
private final Block<T> dictionary;
private final int idsOffset;
private final int[] ids;
protected final int idsOffset;
protected final int[] ids;
private final long retainedSizeInBytes;
private volatile long sizeInBytes = -1;
private volatile long logicalSizeInBytes = -1;

60
presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlockEncoding.java
View File

@ -13,9 +13,16 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.airlift.slice.Slices;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
public class DictionaryBlockEncoding
implements BlockEncoding
@ -76,4 +83,57 @@ public class DictionaryBlockEncoding
// TODO: fix DictionaryBlock so that dictionaryIsCompacted can be set to true when the underlying block over-retains memory.
return new DictionaryBlock(positionCount, dictionaryBlock, ids, false, new DictionaryId(mostSignificantBits, leastSignificantBits, sequenceId));
}
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for Dictionary ReadBlock");
}
Input input = (Input) inputStream;
int positionCount = input.readInt();
// dictionary
Block dictionaryBlock = blockEncodingSerde.readBlock(input);
// ids
int[] ids = input.readInts(positionCount);
// instance id
long mostSignificantBits = input.readLong();
long leastSignificantBits = input.readLong();
long sequenceId = input.readLong();
return new DictionaryBlock(positionCount, dictionaryBlock, ids, false, new DictionaryId(mostSignificantBits, leastSignificantBits, sequenceId));
}
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock");
}
Output output = (Output) outputStream;
DictionaryBlock dictionaryBlock = (DictionaryBlock) block;
dictionaryBlock = dictionaryBlock.compact();
// positionCount
int positionCount = dictionaryBlock.getPositionCount();
output.writeInt(positionCount);
// dictionary
Block dictionary = dictionaryBlock.getDictionary();
blockEncodingSerde.writeBlock(output, dictionary);
// ids
output.writeInts(dictionaryBlock.ids, dictionaryBlock.idsOffset, positionCount);
// instance id
output.writeLong(dictionaryBlock.getDictionarySourceId().getMostSignificantBits());
output.writeLong(dictionaryBlock.getDictionarySourceId().getLeastSignificantBits());
output.writeLong(dictionaryBlock.getDictionarySourceId().getSequenceId());
}
}

6
presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlock.java
View File

@ -38,11 +38,11 @@ public class Int128ArrayBlock
private static final int INSTANCE_SIZE = ClassLayout.parseClass(Int128ArrayBlock.class).instanceSize();
public static final int INT128_BYTES = Long.BYTES + Long.BYTES;
private final int positionOffset;
protected final int positionOffset;
private final int positionCount;
@Nullable
private final boolean[] valueIsNull;
private final long[] values;
protected final boolean[] valueIsNull;
protected final long[] values;
private final long sizeInBytes;
private final long retainedSizeInBytes;

70
presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlockEncoding.java
View File

@ -13,14 +13,22 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.prestosql.spi.block.EncoderUtil.decodeNullBits;
import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits;
public class Int128ArrayBlockEncoding
implements BlockEncoding
extends AbstractBlockEncoding<Int128ArrayBlock>
{
public static final String NAME = "INT128_ARRAY";
@ -63,4 +71,64 @@ public class Int128ArrayBlockEncoding
return new Int128ArrayBlock(0, positionCount, valueIsNull, values);
}
@Override
public void write(Kryo kryo, Output output, Int128ArrayBlock block)
{
int positionCount = block.getPositionCount();
output.writeInt(positionCount);
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(block.valueIsNull, 0, positionCount);
}
output.writeLongs(block.values, block.positionOffset, positionCount * 2);
}
@Override
public Int128ArrayBlock read(Kryo kryo, Input input, Class<? extends Int128ArrayBlock> aClass)
{
int positionCount = input.readInt();
boolean[] valuesIsNull = null;
if (input.readBoolean()) {
valuesIsNull = input.readBooleans(positionCount);
}
long[] values = input.readLongs(positionCount * 2);
return new Int128ArrayBlock(0, positionCount, valuesIsNull, values);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param input
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for Int128");
}
return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, Int128ArrayBlock.class);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param output
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for Int128");
}
this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (Int128ArrayBlock) block);
}
}

12
presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlock.java
View File

@ -35,11 +35,11 @@ public class IntArrayBlock
{
private static final int INSTANCE_SIZE = ClassLayout.parseClass(IntArrayBlock.class).instanceSize();
private final int arrayOffset;
protected final int arrayOffset;
private final int positionCount;
@Nullable
private final boolean[] valueIsNull;
private final int[] values;
protected final boolean[] valueIsNull;
protected final int[] values;
private final long sizeInBytes;
private final long retainedSizeInBytes;
@ -136,12 +136,6 @@ public class IntArrayBlock
return getInt(position, offset);
}
@Override
public String getString(int position, int offset, int length)
{
return String.valueOf(getInt(position, offset));
}
@Override
public boolean mayHaveNull()
{

70
presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlockEncoding.java
View File

@ -13,14 +13,22 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.prestosql.spi.block.EncoderUtil.decodeNullBits;
import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits;
public class IntArrayBlockEncoding
implements BlockEncoding
extends AbstractBlockEncoding<IntArrayBlock>
{
public static final String NAME = "INT_ARRAY";
@ -61,4 +69,64 @@ public class IntArrayBlockEncoding
return new IntArrayBlock(0, positionCount, valueIsNull, values);
}
@Override
public void write(Kryo kryo, Output output, IntArrayBlock block)
{
int positionCount = block.getPositionCount();
output.writeInt(positionCount);
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(block.valueIsNull, 0, positionCount);
}
output.writeInts(block.values, block.arrayOffset, positionCount);
}
@Override
public IntArrayBlock read(Kryo kryo, Input input, Class<? extends IntArrayBlock> aClass)
{
int positionCount = input.readInt();
boolean[] valuesIsNull = null;
if (input.readBoolean()) {
valuesIsNull = input.readBooleans(positionCount);
}
int[] values = input.readInts(positionCount);
return new IntArrayBlock(0, positionCount, valuesIsNull, values);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param input
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for IntArrayBlock");
}
return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, IntArrayBlock.class);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param output
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for IntArrayBlock");
}
this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (IntArrayBlock) block);
}
}

14
presto-spi/src/main/java/io/prestosql/spi/block/LazyBlockEncoding.java
View File

@ -16,6 +16,8 @@ package io.prestosql.spi.block;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Optional;
public class LazyBlockEncoding
@ -45,6 +47,18 @@ public class LazyBlockEncoding
throw new UnsupportedOperationException();
}
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input)
{
throw new UnsupportedOperationException();
}
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block)
{
throw new UnsupportedOperationException();
}
@Override
public Optional<Block> replacementBlockForWrite(Block block)
{

6
presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlock.java
View File

@ -39,11 +39,11 @@ public class LongArrayBlock
//can we intro operations at block level?, for example join of blocks?
private static final int INSTANCE_SIZE = ClassLayout.parseClass(LongArrayBlock.class).instanceSize();
private final int arrayOffset;
protected final int arrayOffset;
private final int positionCount;
@Nullable
private final boolean[] valueIsNull;
private final long[] values; //change to use offheap --> accessible by RDMA
protected final boolean[] valueIsNull;
protected final long[] values; //change to use offheap --> accessible by RDMA
private final long sizeInBytes;
private final long retainedSizeInBytes;

70
presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlockEncoding.java
View File

@ -13,14 +13,22 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.prestosql.spi.block.EncoderUtil.decodeNullBits;
import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits;
public class LongArrayBlockEncoding
implements BlockEncoding
extends AbstractBlockEncoding<LongArrayBlock>
{
public static final String NAME = "LONG_ARRAY";
@ -61,4 +69,64 @@ public class LongArrayBlockEncoding
return new LongArrayBlock(0, positionCount, valueIsNull, values);
}
@Override
public void write(Kryo kryo, Output output, LongArrayBlock block)
{
int positionCount = block.getPositionCount();
output.writeInt(positionCount);
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(block.valueIsNull, 0, positionCount);
}
output.writeLongs(block.values, block.arrayOffset, positionCount);
}
@Override
public LongArrayBlock read(Kryo kryo, Input input, Class<? extends LongArrayBlock> aClass)
{
int positionCount = input.readInt();
boolean[] valuesIsNull = null;
if (input.readBoolean()) {
valuesIsNull = input.readBooleans(positionCount);
}
long[] values = input.readLongs(positionCount);
return new LongArrayBlock(0, positionCount, valuesIsNull, values);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param input
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for LongArrayBlock");
}
return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, LongArrayBlock.class);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param output
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for LongArrayBlock");
}
this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (LongArrayBlock) block);
}
}

75
presto-spi/src/main/java/io/prestosql/spi/block/MapBlockEncoding.java
View File

@ -14,12 +14,18 @@
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import io.prestosql.spi.type.MapType;
import io.prestosql.spi.type.TypeManager;
import io.prestosql.spi.type.TypeSerde;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Optional;
import static io.airlift.slice.Slices.wrappedIntArray;
@ -96,4 +102,73 @@ public class MapBlockEncoding
Optional<boolean[]> mapIsNull = EncoderUtil.decodeNullBits(sliceInput, positionCount);
return createMapBlockInternal(mapType, 0, positionCount, mapIsNull, offsets, keyBlock, valueBlock, hashTable);
}
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for MapBlock ReadBlock");
}
Input input = (Input) inputStream;
MapType mapType = (MapType) TypeSerde.readType(typeManager, input);
Block keyBlock = blockEncodingSerde.readBlock(input);
Block valueBlock = blockEncodingSerde.readBlock(input);
int hashTableSize = input.readInt();
int[] hashTable = input.readInts(hashTableSize);
if (keyBlock.getPositionCount() != valueBlock.getPositionCount() || keyBlock.getPositionCount() * HASH_MULTIPLIER != hashTable.length) {
throw new IllegalArgumentException(
format("Deserialized MapBlock violates invariants: key %d, value %d, hash %d", keyBlock.getPositionCount(), valueBlock.getPositionCount(), hashTable.length));
}
int positionCount = input.readInt();
int[] offsets = input.readInts(positionCount + 1);
Optional<boolean[]> mapIsNull = Optional.empty();
if (input.readBoolean()) {
mapIsNull = Optional.of(input.readBooleans(positionCount));
}
return createMapBlockInternal(mapType, 0, positionCount, mapIsNull, offsets, keyBlock, valueBlock, hashTable);
}
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock");
}
Output output = (Output) outputStream;
AbstractMapBlock mapBlock = (AbstractMapBlock) block;
int positionCount = mapBlock.getPositionCount();
int offsetBase = mapBlock.getOffsetBase();
int[] offsets = mapBlock.getOffsets();
int[] hashTable = mapBlock.getHashTables();
int entriesStartOffset = offsets[offsetBase];
int entriesEndOffset = offsets[offsetBase + positionCount];
TypeSerde.writeType(output, mapBlock.mapType);
blockEncodingSerde.writeBlock(output, mapBlock.getRawKeyBlock().getRegion(entriesStartOffset, entriesEndOffset - entriesStartOffset));
blockEncodingSerde.writeBlock(output, mapBlock.getRawValueBlock().getRegion(entriesStartOffset, entriesEndOffset - entriesStartOffset));
output.writeInt((entriesEndOffset - entriesStartOffset) * HASH_MULTIPLIER);
output.writeInts(hashTable, entriesStartOffset * HASH_MULTIPLIER, (entriesEndOffset - entriesStartOffset) * HASH_MULTIPLIER);
output.writeInt(positionCount);
for (int position = 0; position < positionCount + 1; position++) {
output.writeInt(offsets[offsetBase + position] - entriesStartOffset);
}
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(mapBlock.getMapIsNull(), mapBlock.getOffsetBase(), positionCount);
}
}
}

69
presto-spi/src/main/java/io/prestosql/spi/block/RowBlockEncoding.java
View File

@ -14,8 +14,15 @@
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.airlift.slice.Slices.wrappedIntArray;
import static io.prestosql.spi.block.RowBlock.createRowBlockInternal;
@ -71,4 +78,66 @@ public class RowBlockEncoding
boolean[] rowIsNull = EncoderUtil.decodeNullBits(sliceInput, positionCount).orElseGet(() -> new boolean[positionCount]);
return createRowBlockInternal(0, positionCount, rowIsNull, fieldBlockOffsets, fieldBlocks);
}
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleRow ReadBlock");
}
Input input = (Input) inputStream;
int numFields = input.readInt();
Block[] fieldBlocks = new Block[numFields];
for (int i = 0; i < fieldBlocks.length; i++) {
fieldBlocks[i] = blockEncodingSerde.readBlock(input);
}
int positionCount = input.readInt();
int[] fieldBlockOffsets = input.readInts(positionCount);
boolean[] rowIsNull;
if (input.readBoolean()) {
rowIsNull = input.readBooleans(positionCount);
}
else {
rowIsNull = new boolean[positionCount];
}
return createRowBlockInternal(0, positionCount, rowIsNull, fieldBlockOffsets, fieldBlocks);
}
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for RowBlock WriteBlock");
}
Output output = (Output) outputStream;
AbstractRowBlock rowBlock = (AbstractRowBlock) block;
int numFields = rowBlock.numFields;
int positionCount = rowBlock.getPositionCount();
int offsetBase = rowBlock.getOffsetBase();
int[] fieldBlockOffsets = rowBlock.getFieldBlockOffsets();
int startFieldBlockOffset = fieldBlockOffsets[offsetBase];
int endFieldBlockOffset = fieldBlockOffsets[offsetBase + positionCount];
output.writeInt(numFields);
for (int i = 0; i < numFields; i++) {
blockEncodingSerde.writeBlock(output, rowBlock.getRawFieldBlocks()[i].getRegion(startFieldBlockOffset, endFieldBlockOffset - startFieldBlockOffset));
}
output.writeInt(positionCount);
for (int position = 0; position < positionCount + 1; position++) {
output.writeInt(fieldBlockOffsets[offsetBase + position] - startFieldBlockOffset);
}
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(rowBlock.getRowIsNull(), offsetBase, positionCount);
}
}
}

52
presto-spi/src/main/java/io/prestosql/spi/block/RunLengthBlockEncoding.java
View File

@ -13,8 +13,15 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
public class RunLengthBlockEncoding
implements BlockEncoding
@ -50,4 +57,49 @@ public class RunLengthBlockEncoding
return new RunLengthEncodedBlock(value, positionCount);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param inputStream
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for RLE ReadBlock");
}
Input input = (Input) inputStream;
int positionCount = input.readInt();
Block value = blockEncodingSerde.readBlock(input);
return new RunLengthEncodedBlock(value, positionCount);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param outputStream
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for RLE WriteBlock");
}
RunLengthEncodedBlock rleBlock = (RunLengthEncodedBlock) block;
Output output = (Output) outputStream;
// write the run length
output.writeInt(rleBlock.getPositionCount());
// write the value
blockEncodingSerde.writeBlock(output, rleBlock.getValue());
}
}

6
presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlock.java
View File

@ -35,11 +35,11 @@ public class ShortArrayBlock
{
private static final int INSTANCE_SIZE = ClassLayout.parseClass(ShortArrayBlock.class).instanceSize();
private final int arrayOffset;
protected final int arrayOffset;
private final int positionCount;
@Nullable
private final boolean[] valueIsNull;
private final short[] values;
protected final boolean[] valueIsNull;
protected final short[] values;
private final long sizeInBytes;
private final long retainedSizeInBytes;

70
presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlockEncoding.java
View File

@ -13,14 +13,22 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.prestosql.spi.block.EncoderUtil.decodeNullBits;
import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits;
public class ShortArrayBlockEncoding
implements BlockEncoding
extends AbstractBlockEncoding<ShortArrayBlock>
{
public static final String NAME = "SHORT_ARRAY";
@ -61,4 +69,64 @@ public class ShortArrayBlockEncoding
return new ShortArrayBlock(0, positionCount, valueIsNull, values);
}
@Override
public void write(Kryo kryo, Output output, ShortArrayBlock block)
{
int positionCount = block.getPositionCount();
output.writeInt(positionCount);
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(block.valueIsNull, 0, positionCount);
}
output.writeShorts(block.values, block.arrayOffset, positionCount);
}
@Override
public ShortArrayBlock read(Kryo kryo, Input input, Class<? extends ShortArrayBlock> aClass)
{
int positionCount = input.readInt();
boolean[] valuesIsNull = null;
if (input.readBoolean()) {
valuesIsNull = input.readBooleans(positionCount);
}
short[] values = input.readShorts(positionCount);
return new ShortArrayBlock(0, positionCount, valuesIsNull, values);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param input
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for ShortArrayBlock");
}
return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, ShortArrayBlock.class);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param output
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for ShortArrayBlock");
}
this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (ShortArrayBlock) block);
}
}

68
presto-spi/src/main/java/io/prestosql/spi/block/SingleMapBlockEncoding.java
View File

@ -14,12 +14,19 @@
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import io.prestosql.spi.type.MapType;
import io.prestosql.spi.type.TypeManager;
import io.prestosql.spi.type.TypeSerde;
import java.io.InputStream;
import java.io.OutputStream;
import static io.airlift.slice.Slices.wrappedIntArray;
import static io.prestosql.spi.block.AbstractMapBlock.HASH_MULTIPLIER;
import static java.lang.String.format;
@ -76,4 +83,65 @@ public class SingleMapBlockEncoding
return new SingleMapBlock(mapType, 0, keyBlock.getPositionCount() * 2, keyBlock, valueBlock, hashTable);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param inputStream
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleMap ReadBlock");
}
Input input = (Input) inputStream;
MapType mapType = (MapType) TypeSerde.readType(typeManager, input);
Block keyBlock = blockEncodingSerde.readBlock(input);
Block valueBlock = blockEncodingSerde.readBlock(input);
int hashTableSize = input.readInt();
int[] hashTable = input.readInts(hashTableSize);
if (keyBlock.getPositionCount() != valueBlock.getPositionCount()
|| keyBlock.getPositionCount() * HASH_MULTIPLIER != hashTable.length) {
throw new IllegalArgumentException(
format("Deserialized SingleMapBlock violates invariants: key %d, value %d, hash %d", keyBlock.getPositionCount(), valueBlock.getPositionCount(), hashTable.length));
}
return new SingleMapBlock(mapType, 0, keyBlock.getPositionCount() * 2, keyBlock, valueBlock, hashTable);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param outputStream
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock");
}
Output output = (Output) outputStream;
SingleMapBlock singleMapBlock = (SingleMapBlock) block;
TypeSerde.writeType(output, singleMapBlock.mapType);
int offset = singleMapBlock.getOffset();
int positionCount = singleMapBlock.getPositionCount();
blockEncodingSerde.writeBlock(output, singleMapBlock.getRawKeyBlock().getRegion(offset / 2, positionCount / 2));
blockEncodingSerde.writeBlock(output, singleMapBlock.getRawValueBlock().getRegion(offset / 2, positionCount / 2));
int[] hashTable = singleMapBlock.getHashTable();
output.writeInt(positionCount / 2 * HASH_MULTIPLIER);
output.writeInts(hashTable, offset / 2 * HASH_MULTIPLIER, positionCount / 2 * HASH_MULTIPLIER);
}
}

56
presto-spi/src/main/java/io/prestosql/spi/block/SingleRowBlockEncoding.java
View File

@ -14,8 +14,15 @@
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
public class SingleRowBlockEncoding
implements BlockEncoding
@ -50,4 +57,53 @@ public class SingleRowBlockEncoding
}
return new SingleRowBlock(0, fieldBlocks);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param inputStream
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleRow ReadBlock");
}
Input input = (Input) inputStream;
int numFields = input.readInt();
Block[] fieldBlocks = new Block[numFields];
for (int i = 0; i < fieldBlocks.length; i++) {
fieldBlocks[i] = blockEncodingSerde.readBlock(input);
}
return new SingleRowBlock(0, fieldBlocks);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param outputStream
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleRowBlock WriteBlock");
}
Output output = (Output) outputStream;
SingleRowBlock singleRowBlock = (SingleRowBlock) block;
int numFields = singleRowBlock.getNumFields();
int rowIndex = singleRowBlock.getRowIndex();
output.writeInt(numFields);
for (int i = 0; i < numFields; i++) {
blockEncodingSerde.writeBlock(output, singleRowBlock.getRawFieldBlock(i).getRegion(rowIndex, 1));
}
}
}

68
presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlock.java
View File

@ -13,9 +13,14 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.Slice;
import io.airlift.slice.SliceOutput;
import io.airlift.slice.Slices;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.util.BloomFilter;
import org.openjdk.jol.info.ClassLayout;
@ -28,6 +33,7 @@ import java.util.function.BiConsumer;
import java.util.function.Function;
import static io.airlift.slice.SizeOf.sizeOf;
import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static io.prestosql.spi.block.BlockUtil.checkArrayRange;
import static io.prestosql.spi.block.BlockUtil.checkValidRegion;
import static io.prestosql.spi.block.BlockUtil.compactArray;
@ -36,18 +42,25 @@ import static io.prestosql.spi.block.BlockUtil.compactSlice;
public class VariableWidthBlock
extends AbstractVariableWidthBlock<byte[]>
implements KryoSerializable
{
private static final int INSTANCE_SIZE = ClassLayout.parseClass(VariableWidthBlock.class).instanceSize();
private final int arrayOffset;
private final int positionCount;
private final Slice slice;
private final int[] offsets;
protected int arrayOffset;
private int positionCount;
private Slice slice;
protected int[] offsets;
@Nullable
private final boolean[] valueIsNull;
protected boolean[] valueIsNull;
private final long retainedSizeInBytes;
private final long sizeInBytes;
private long retainedSizeInBytes;
private long sizeInBytes;
private boolean isInitialized;
public VariableWidthBlock()
{
arrayOffset = 0;
}
public VariableWidthBlock(int positionCount, Slice slice, int[] offsets, Optional<boolean[]> valueIsNull)
{
@ -82,6 +95,8 @@ public class VariableWidthBlock
sizeInBytes = offsets[arrayOffset + positionCount] - offsets[arrayOffset] + ((Integer.BYTES + Byte.BYTES) * (long) positionCount);
retainedSizeInBytes = INSTANCE_SIZE + slice.getRetainedSize() + sizeOf(valueIsNull) + sizeOf(offsets);
this.isInitialized = true;
}
@Override
@ -297,4 +312,43 @@ public class VariableWidthBlock
{
return Objects.hash(arrayOffset, positionCount, slice, Arrays.hashCode(offsets), Arrays.hashCode(valueIsNull), retainedSizeInBytes, sizeInBytes);
}
@Override
public void write(Kryo kryo, Output output)
{
/* # of positions
* [Length Per Position]
* [nulls as bits]
* [buffer]
*/
output.write(getPositionCount());
output.writeInts(offsets, arrayOffset, positionCount + 1);
output.writeBoolean(mayHaveNull());
if (mayHaveNull()) {
output.writeBooleans(valueIsNull, 0, positionCount);
}
output.write(offsets[arrayOffset + positionCount] - offsets[arrayOffset]);
output.write(slice.byteArray(), offsets[arrayOffset],
offsets[arrayOffset + positionCount] - offsets[arrayOffset]);
}
@Override
public void read(Kryo kryo, Input input)
{
if (isInitialized) {
throw new PrestoException(GENERIC_INTERNAL_ERROR, "Already initialized block");
}
positionCount = input.read();
offsets = input.readInts(positionCount + 1);
if (input.readBoolean()) {
valueIsNull = input.readBooleans(positionCount);
}
int blockSize = input.read();
slice = Slices.wrappedBuffer(input.readBytes(blockSize));
isInitialized = true;
sizeInBytes = offsets[arrayOffset + positionCount] - offsets[arrayOffset] + ((Integer.BYTES + Byte.BYTES) * (long) positionCount);
retainedSizeInBytes = INSTANCE_SIZE + slice.getRetainedSize() + sizeOf(valueIsNull) + sizeOf(offsets);
}
}

76
presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlockEncoding.java
View File

@ -13,17 +13,25 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.Slice;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.airlift.slice.Slices;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import static io.airlift.slice.SizeOf.SIZE_OF_INT;
import static io.prestosql.spi.block.EncoderUtil.decodeNullBits;
import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits;
public class VariableWidthBlockEncoding
implements BlockEncoding
extends AbstractBlockEncoding<VariableWidthBlock>
{
public static final String NAME = "VARIABLE_WIDTH";
@ -72,4 +80,70 @@ public class VariableWidthBlockEncoding
return new VariableWidthBlock(0, positionCount, slice, offsets, valueIsNull);
}
@Override
public void write(Kryo kryo, Output output, VariableWidthBlock block)
{
int positionCount = block.getPositionCount();
output.writeInt(positionCount);
output.writeInts(block.offsets, block.arrayOffset, positionCount + 1);
output.writeBoolean(block.mayHaveNull());
if (block.mayHaveNull()) {
output.writeBooleans(block.valueIsNull, 0, positionCount);
}
int totalSize = block.offsets[block.arrayOffset + positionCount] - block.offsets[block.arrayOffset];
output.writeInt(totalSize);
output.write(block.getRawSlice(0).byteArray(), block.offsets[block.arrayOffset], totalSize);
}
@Override
public VariableWidthBlock read(Kryo kryo, Input input, Class<? extends VariableWidthBlock> aClass)
{
int positionCount = input.readInt();
int[] offsets = input.readInts(positionCount + 1);
boolean[] valuesIsNull = null;
if (input.readBoolean()) {
valuesIsNull = input.readBooleans(positionCount);
}
int blockSize = input.readInt();
Slice slice = Slices.wrappedBuffer(input.readBytes(blockSize));
return new VariableWidthBlock(0, positionCount, slice, offsets, valuesIsNull);
}
/**
* Read a block from the specified input. The returned
* block should begin at the specified position.
*
* @param blockEncodingSerde
* @param input
*/
@Override
public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for VariableWidthBlock");
}
return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, VariableWidthBlock.class);
}
/**
* Write the specified block to the specified output
*
* @param blockEncodingSerde
* @param output
* @param block
*/
@Override
public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block)
{
if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) {
throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for VariableWidthBlock");
}
this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (VariableWidthBlock) block);
}
}

35
presto-spi/src/main/java/io/prestosql/spi/type/TypeSerde.java
View File

@ -13,6 +13,8 @@
*/
package io.prestosql.spi.type;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
@ -59,4 +61,37 @@ public final class TypeSerde
output.writeInt(bytes.length);
output.writeBytes(bytes);
}
public static void writeType(Output output, Type type)
{
requireNonNull(output, "output is null");
requireNonNull(type, "type is null");
writeLengthPrefixedString(output, type.getTypeSignature().toString());
}
public static Type readType(TypeManager typeManager, Input sliceInput)
{
requireNonNull(sliceInput, "sliceInput is null");
String name = readLengthPrefixedString(sliceInput);
Type type = typeManager.getType(parseTypeSignature(name));
if (type == null) {
throw new IllegalArgumentException("Unknown type " + name);
}
return type;
}
private static String readLengthPrefixedString(Input input)
{
int length = input.readInt();
byte[] bytes = input.readBytes(length);
return new String(bytes, UTF_8);
}
private static void writeLengthPrefixedString(Output output, String string)
{
byte[] bytes = string.getBytes(UTF_8);
output.writeInt(bytes.length);
output.writeBytes(bytes);
}
}

113
presto-spi/src/test/java/io/prestosql/spi/block/TestInt128ArrayBlockEncoding.java Normal file
View File

@ -0,0 +1,113 @@
/*
* Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.spi.block;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.InputStreamSliceInput;
import io.airlift.slice.OutputStreamSliceOutput;
import io.prestosql.spi.type.Type;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Optional;
import static io.prestosql.spi.block.TestingSession.SESSION;
import static io.prestosql.spi.type.BigintType.BIGINT;
import static org.testng.Assert.assertEquals;
public class TestInt128ArrayBlockEncoding
{
private final BlockEncodingSerde blockEncodingSerde = new TestingBlockEncodingSerde();
private String storePath = "./target/store";
@BeforeClass
public void init()
{
File dir = new File(storePath);
if (!dir.exists()) {
dir.mkdirs();
}
}
@AfterClass
public void teardown()
{
File dir = new File(storePath);
if (dir.exists()) {
File[] files = dir.listFiles();
for (File file : files) {
file.delete();
}
dir.delete();
}
}
@Test
public void testRoundTrip() throws IOException
{
int count = 1024;
Int128ArrayBlock expectedBlock = new Int128ArrayBlock(count, Optional.empty(), getValues(count * 2));
OutputStreamSliceOutput sliceOutput = new OutputStreamSliceOutput(new FileOutputStream(storePath + "/" + "sliceFile.dat"));
InputStreamSliceInput sliceInput = new InputStreamSliceInput(new FileInputStream(storePath + "/" + "sliceFile.dat"));
blockEncodingSerde.writeBlock(sliceOutput, expectedBlock);
sliceOutput.close();
Block actualBlock = blockEncodingSerde.readBlock(sliceInput);
sliceInput.close();
assertBlockEquals(BIGINT, actualBlock, expectedBlock);
}
@Test
public void testRoundTripDirect() throws FileNotFoundException
{
int count = 1024;
Int128ArrayBlock expectedBlock = new Int128ArrayBlock(count, Optional.empty(), getValues(count * 2));
Output output = new Output(new FileOutputStream(storePath + "/" + "file.dat"));
Input input = new Input(new FileInputStream(storePath + "/" + "file.dat"));
blockEncodingSerde.writeBlock(output, expectedBlock);
output.close();
Block actualBlock = blockEncodingSerde.readBlock(input);
input.close();
assertBlockEquals(BIGINT, actualBlock, expectedBlock);
}
private long[] getValues(int count)
{
long[] values = new long[count];
for (int i = 0; i < values.length; i++) {
values[i] = i;
}
return values;
}
private static void assertBlockEquals(Type type, Block actual, Block expected)
{
for (int position = 0; position < actual.getPositionCount(); position++) {
assertEquals(type.getObjectValue(SESSION, actual, position), type.getObjectValue(SESSION, expected, position));
}
}
}

125
presto-spi/src/test/java/io/prestosql/spi/block/TestVariableWidthBlockEncoding.java
View File

@ -13,10 +13,25 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.common.base.Stopwatch;
import io.airlift.slice.DynamicSliceOutput;
import io.airlift.slice.InputStreamSliceInput;
import io.airlift.slice.OutputStreamSliceOutput;
import io.prestosql.spi.type.Type;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import static io.prestosql.spi.block.TestingSession.SESSION;
import static io.prestosql.spi.type.VarcharType.VARCHAR;
import static org.testng.Assert.assertEquals;
@ -24,6 +39,36 @@ import static org.testng.Assert.assertEquals;
public class TestVariableWidthBlockEncoding
{
private final BlockEncodingSerde blockEncodingSerde = new TestingBlockEncodingSerde();
private String storePath = "./target/store";
private Kryo kryo;
private Output output;
private Input input;
@BeforeClass
public void init()
{
kryo = new Kryo();
kryo.register(VariableWidthBlock.class);
File dir = new File(storePath);
if (!dir.exists()) {
dir.mkdirs();
}
}
@AfterClass
public void teardown()
{
File dir = new File(storePath);
if (dir.exists()) {
File[] files = dir.listFiles();
for (File file : files) {
file.delete();
}
dir.delete();
}
}
@Test
public void testRoundTrip()
@ -41,6 +86,86 @@ public class TestVariableWidthBlockEncoding
assertBlockEquals(VARCHAR, actualBlock, expectedBlock);
}
@Test
public void testRoundTripKryo() throws FileNotFoundException
{
output = new Output(new FileOutputStream(storePath + "/" + "file.dat"));
input = new Input(new FileInputStream(storePath + "/" + "file.dat"));
BlockBuilder expectedBlockBuilder = VARCHAR.createBlockBuilder(null, 4);
VARCHAR.writeString(expectedBlockBuilder, "alice");
VARCHAR.writeString(expectedBlockBuilder, "bob");
VARCHAR.writeString(expectedBlockBuilder, "charlie");
VARCHAR.writeString(expectedBlockBuilder, "dave");
Block expectedBlock = expectedBlockBuilder.build();
kryo.writeObject(output, expectedBlock);
output.close();
Block actualBlock = kryo.readObject(input, VariableWidthBlock.class);
assertBlockEquals(VARCHAR, actualBlock, expectedBlock);
input.close();
}
public void testRoundTripKryoPerf100000000() throws IOException
{
int loopCount = 1000;
for (int i = 1; i <= 5; i++) {
loopCount *= 10;
loopReadWritePerfTest(loopCount);
}
}
private void loopReadWritePerfTest(int loopCount) throws IOException
{
output = new Output(new FileOutputStream(storePath + "/" + "file.dat"));
input = new Input(new FileInputStream(storePath + "/" + "file.dat"));
OutputStreamSliceOutput sliceOutput = new OutputStreamSliceOutput(new FileOutputStream(storePath + "/" + "sliceFile.dat"));
InputStreamSliceInput sliceInput = new InputStreamSliceInput(new FileInputStream(storePath + "/" + "sliceFile.dat"));
BlockBuilder expectedBlockBuilder = VARCHAR.createBlockBuilder(null, 4);
VARCHAR.writeString(expectedBlockBuilder, "alice");
VARCHAR.writeString(expectedBlockBuilder, "bob");
VARCHAR.writeString(expectedBlockBuilder, "charlie");
VARCHAR.writeString(expectedBlockBuilder, "dave");
Block expectedBlock = expectedBlockBuilder.build();
Stopwatch watchKryoWrite = Stopwatch.createStarted();
for (int i = 0; i < loopCount; i++) {
kryo.writeObject(output, expectedBlock);
}
watchKryoWrite.stop();
System.out.println(String.format("[Pages: %,11d] Time to write [Kryo]: %,7d ms", loopCount, watchKryoWrite.elapsed(TimeUnit.MILLISECONDS)));
output.close();
Stopwatch watchSerDeWrite = Stopwatch.createStarted();
for (int i = 0; i < loopCount; i++) {
blockEncodingSerde.writeBlock(sliceOutput, expectedBlock);
}
watchSerDeWrite.stop();
System.out.println(String.format("[Pages: %,11d] Time to write [BlockSerDe]: %,7d ms", loopCount, watchSerDeWrite.elapsed(TimeUnit.MILLISECONDS)));
sliceOutput.close();
Stopwatch watchKryoRead = Stopwatch.createStarted();
for (int i = 0; i < loopCount; i++) {
Block actualBlock = kryo.readObject(input, VariableWidthBlock.class);
assertBlockEquals(VARCHAR, actualBlock, expectedBlock);
}
watchKryoRead.stop();
System.out.println(String.format("[Pages: %,11d] Time to read [Kryo]: %,7d ms", loopCount, watchKryoRead.elapsed(TimeUnit.MILLISECONDS)));
input.close();
Stopwatch watchSerDeRead = Stopwatch.createStarted();
for (int i = 0; i < loopCount; i++) {
Block actualBlock = blockEncodingSerde.readBlock(sliceInput);
assertBlockEquals(VARCHAR, actualBlock, expectedBlock);
}
watchSerDeRead.stop();
System.out.println(String.format("[Pages: %,11d] Time to read [BlockSerDe]: %,7d ms", loopCount, watchSerDeRead.elapsed(TimeUnit.MILLISECONDS)));
input.close();
}
private static void assertBlockEquals(Type type, Block actual, Block expected)
{
for (int position = 0; position < actual.getPositionCount(); position++) {

78
presto-spi/src/test/java/io/prestosql/spi/block/TestingBlockEncodingSerde.java
View File

@ -13,14 +13,23 @@
*/
package io.prestosql.spi.block;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.airlift.slice.SliceInput;
import io.airlift.slice.SliceOutput;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static com.google.common.base.Preconditions.checkArgument;
import static io.prestosql.spi.StandardErrorCode.TYPE_NOT_FOUND;
import static java.nio.charset.StandardCharsets.UTF_8;
// This class is exactly the same as BlockEncodingManager. They are in SPI and don't have access to InternalBlockEncodingSerde.
@ -28,6 +37,7 @@ public final class TestingBlockEncodingSerde
implements BlockEncodingSerde
{
private final ConcurrentMap<String, BlockEncoding> blockEncodings = new ConcurrentHashMap<>();
private final Kryo kryo = new Kryo();
public TestingBlockEncodingSerde()
{
@ -92,6 +102,74 @@ public final class TestingBlockEncodingSerde
}
}
@Override
public Block readBlock(InputStream inputStream)
{
if (!(inputStream instanceof Input)) {
throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR,
"This interface should not be called in this flow");
}
Input input = (Input) inputStream;
String encodingName;
// read the encoding name
encodingName = readLengthPrefixedString((Input) input);
// look up the encoding factory
BlockEncoding blockEncoding = blockEncodings.get(encodingName);
Serializer<?> serializer = getSerializerFromBlockEncoding(blockEncoding);
if (serializer == null) {
throw new PrestoException(TYPE_NOT_FOUND, "BlockEncoding Type not implemented: " + blockEncoding);
}
return (Block) serializer.read(kryo, input, null);
}
@Override
public void writeBlock(OutputStream outputStream, Block block)
{
if (!(outputStream instanceof Output)) {
throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR,
"This interface should not be called in this flow");
}
Output output = (Output) outputStream;
String encodingName = block.getEncodingName();
BlockEncoding blockEncoding = blockEncodings.get(encodingName);
Serializer<Block<?>> serializer = getSerializerFromBlockEncoding(blockEncoding);
if (serializer == null) {
throw new PrestoException(TYPE_NOT_FOUND, "BlockEncoding Type not implemented: " + blockEncoding);
}
// write the name to the output
writeLengthPrefixedString(output, encodingName);
// write the block to the output
serializer.write(kryo, output, block);
}
private Serializer<Block<?>> getSerializerFromBlockEncoding(BlockEncoding blockEncoding)
{
if (blockEncoding instanceof AbstractBlockEncoding) {
return (Serializer<Block<?>>) blockEncoding;
}
return null;
}
private static String readLengthPrefixedString(Input input)
{
int length = input.readInt();
byte[] bytes = input.readBytes(length);
return new String(bytes, UTF_8);
}
private static void writeLengthPrefixedString(Output output, String value)
{
byte[] bytes = value.getBytes(UTF_8);
output.writeInt(bytes.length);
output.writeBytes(bytes);
}
private static String readLengthPrefixedString(SliceInput input)
{
int length = input.readInt();