Add support for vectorized null suppression for block serde

core/trino-main/pom.xml

@@ -556,4 +556,29 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
+
+    <build>
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-compiler-plugin</artifactId>
+                    <configuration>
+                        <!-- Ensure incubator Vector API is on the module path for javac -->
+                        <compilerArgs combine.self="merge">
+                            <arg>${extraJavaVectorArgs}</arg>
+                        </compilerArgs>
+                    </configuration>
+                </plugin>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-javadoc-plugin</artifactId>
+                    <configuration>
+                        <!-- Ensure javadoc resolves incubator Vector API -->
+                        <additionalOptions combine.self="merge">${extraJavaVectorArgs}</additionalOptions>
+                    </configuration>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+    </build>
 </project>

core/trino-main/src/main/java/io/trino/FeaturesConfig.java

@@ -94,6 +94,7 @@ public class FeaturesConfig
      * default value is overwritten for fault tolerant execution in {@link #applyFaultTolerantExecutionDefaults()}}
      */
     private CompressionCodec exchangeCompressionCodec = NONE;
+    private BlockSerdeVectorizedNullSuppressionStrategy blockSerdeVectorizedNullSuppressionStrategy = BlockSerdeVectorizedNullSuppressionStrategy.AUTO;
     private boolean pagesIndexEagerCompactionEnabled;
     private boolean omitDateTimeTypePrecision;
     private int maxRecursionDepth = 10;
@@ -133,6 +134,12 @@ public enum DataIntegrityVerification
         /**/;
     }
 
+    public enum BlockSerdeVectorizedNullSuppressionStrategy
+    {
+        AUTO,
+        NONE,
+    }
+
     public boolean isOmitDateTimeTypePrecision()
     {
         return omitDateTimeTypePrecision;
@@ -366,6 +373,19 @@ public FeaturesConfig setExchangeCompressionCodec(CompressionCodec exchangeCompr
         return this;
     }
 
+    @Config("experimental.blockserde-vectorized-null-suppression-strategy")
+    @ConfigDescription("Strategy used for vectorized null suppression in block serde")
+    public FeaturesConfig setBlockSerdeVectorizedNullSuppressionStrategy(BlockSerdeVectorizedNullSuppressionStrategy blockSerdeVectorizedNullSuppressionStrategy)
+    {
+        this.blockSerdeVectorizedNullSuppressionStrategy = blockSerdeVectorizedNullSuppressionStrategy;
+        return this;
+    }
+
+    public BlockSerdeVectorizedNullSuppressionStrategy getBlockSerdeVectorizedNullSuppressionStrategy()
+    {
+        return blockSerdeVectorizedNullSuppressionStrategy;
+    }
+
     public DataIntegrityVerification getExchangeDataIntegrityVerification()
     {
         return exchangeDataIntegrityVerification;

core/trino-main/src/main/java/io/trino/metadata/BlockEncodingManager.java

@@ -13,6 +13,8 @@
  */
 package io.trino.metadata;
 
+import com.google.inject.Inject;
+import io.trino.simd.BlockEncodingSimdSupport;
 import io.trino.spi.block.ArrayBlockEncoding;
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockEncoding;
@@ -32,6 +34,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static io.trino.simd.BlockEncodingSimdSupport.TESTING_BLOCK_ENCODING_SIMD_SUPPORT;
 import static java.util.Objects.requireNonNull;
 
 public final class BlockEncodingManager
@@ -41,14 +44,19 @@ public final class BlockEncodingManager
     // for serialization
     private final Map<Class<? extends Block>, BlockEncoding> blockEncodingNamesByClass = new ConcurrentHashMap<>();
 
-    public BlockEncodingManager()
+    public static final BlockEncodingManager TESTING_BLOCK_ENCODING_MANAGER = new BlockEncodingManager(TESTING_BLOCK_ENCODING_SIMD_SUPPORT);
+
+    @Inject
+    public BlockEncodingManager(
+            BlockEncodingSimdSupport blockEncodingSimdSupport)
     {
         // add the built-in BlockEncodings
+        BlockEncodingSimdSupport.SimdSupport simdSupport = blockEncodingSimdSupport.getSimdSupport();
         addBlockEncoding(new VariableWidthBlockEncoding());
-        addBlockEncoding(new ByteArrayBlockEncoding());
-        addBlockEncoding(new ShortArrayBlockEncoding());
-        addBlockEncoding(new IntArrayBlockEncoding());
-        addBlockEncoding(new LongArrayBlockEncoding());
+        addBlockEncoding(new ByteArrayBlockEncoding(simdSupport.expandAndCompressByte()));
+        addBlockEncoding(new ShortArrayBlockEncoding(simdSupport.expandAndCompressShort()));
+        addBlockEncoding(new IntArrayBlockEncoding(simdSupport.expandAndCompressInt()));
+        addBlockEncoding(new LongArrayBlockEncoding(simdSupport.expandAndCompressLong()));
         addBlockEncoding(new Fixed12BlockEncoding());
         addBlockEncoding(new Int128ArrayBlockEncoding());
         addBlockEncoding(new DictionaryBlockEncoding());

core/trino-main/src/main/java/io/trino/server/ServerMainModule.java

@@ -100,6 +100,7 @@
 import io.trino.server.protocol.PreparedStatementEncoder;
 import io.trino.server.protocol.spooling.SpoolingServerModule;
 import io.trino.server.remotetask.HttpLocationFactory;
+import io.trino.simd.BlockEncodingSimdSupport;
 import io.trino.spi.PageIndexerFactory;
 import io.trino.spi.PageSorter;
 import io.trino.spi.VersionEmbedder;
@@ -427,6 +428,9 @@ protected void setup(Binder binder)
                 .to(ServerPluginsProvider.class).in(Scopes.SINGLETON);
         configBinder(binder).bindConfig(ServerPluginsProviderConfig.class);
 
+        // SIMD support
+        binder.bind(BlockEncodingSimdSupport.class).in(Scopes.SINGLETON);
+
         // block encodings
         binder.bind(BlockEncodingManager.class).in(Scopes.SINGLETON);
         jsonBinder(binder).addSerializerBinding(Block.class).to(BlockJsonSerde.Serializer.class);

core/trino-main/src/main/java/io/trino/simd/BlockEncodingSimdSupport.java

@@ -0,0 +1,112 @@
+/*
+ * 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.trino.simd;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.trino.FeaturesConfig;
+import io.trino.util.MachineInfo;
+import jdk.incubator.vector.ByteVector;
+import jdk.incubator.vector.IntVector;
+import jdk.incubator.vector.LongVector;
+import jdk.incubator.vector.ShortVector;
+import oshi.hardware.CentralProcessor.ProcessorIdentifier;
+
+import java.util.EnumSet;
+import java.util.Set;
+
+import static io.trino.FeaturesConfig.BlockSerdeVectorizedNullSuppressionStrategy.AUTO;
+import static io.trino.util.MachineInfo.readCpuFlags;
+import static java.util.Locale.ENGLISH;
+
+@Singleton
+public final class BlockEncodingSimdSupport
+{
+    public record SimdSupport(
+            boolean expandAndCompressByte,
+            boolean expandAndCompressShort,
+            boolean expandAndCompressInt,
+            boolean expandAndCompressLong)
+    {
+        public static final SimdSupport NONE = new SimdSupport(false, false, false, false);
+        public static final SimdSupport ALL = new SimdSupport(true, true, true, true);
+    }
+
+    public static final int MINIMUM_SIMD_LENGTH = 512;
+    private final SimdSupport simdSupport;
+    private static final SimdSupport AUTO_DETECTED_SUPPORT = detectSimd();
+
+    public static final BlockEncodingSimdSupport TESTING_BLOCK_ENCODING_SIMD_SUPPORT = new BlockEncodingSimdSupport(true);
+
+    @Inject
+    public BlockEncodingSimdSupport(
+            FeaturesConfig featuresConfig)
+    {
+        this(featuresConfig.getBlockSerdeVectorizedNullSuppressionStrategy().equals(AUTO));
+    }
+
+    public BlockEncodingSimdSupport(
+            boolean enableAutoDetectedSimdSupport)
+    {
+        if (enableAutoDetectedSimdSupport) {
+            simdSupport = AUTO_DETECTED_SUPPORT;
+        }
+        else {
+            simdSupport = SimdSupport.NONE;
+        }
+    }
+
+    private static SimdSupport detectSimd()
+    {
+        ProcessorIdentifier id = MachineInfo.getProcessorInfo();
+
+        String vendor = id.getVendor().toLowerCase(ENGLISH);
+
+        if (vendor.contains("intel") || vendor.contains("amd")) {
+            return detectX86SimdSupport();
+        }
+
+        return SimdSupport.NONE;
+    }
+
+    private static SimdSupport detectX86SimdSupport()
+    {
+        enum X86SimdInstructionSet {
+            avx512f,
+            avx512vbmi2
+        }
+
+        Set<String> flags = readCpuFlags();
+        EnumSet<X86SimdInstructionSet> x86Flags = EnumSet.noneOf(X86SimdInstructionSet.class);
+
+        if (!flags.isEmpty()) {
+            for (X86SimdInstructionSet instructionSet : X86SimdInstructionSet.values()) {
+                if (flags.contains(instructionSet.name())) {
+                    x86Flags.add(instructionSet);
+                }
+            }
+        }
+
+        return new SimdSupport(
+                (ByteVector.SPECIES_PREFERRED.vectorBitSize() >= MINIMUM_SIMD_LENGTH) && x86Flags.contains(X86SimdInstructionSet.avx512vbmi2),
+                (ShortVector.SPECIES_PREFERRED.vectorBitSize() >= MINIMUM_SIMD_LENGTH) && x86Flags.contains(X86SimdInstructionSet.avx512vbmi2),
+                (IntVector.SPECIES_PREFERRED.vectorBitSize() >= MINIMUM_SIMD_LENGTH) && x86Flags.contains(X86SimdInstructionSet.avx512f),
+                (LongVector.SPECIES_PREFERRED.vectorBitSize() >= MINIMUM_SIMD_LENGTH) && x86Flags.contains(X86SimdInstructionSet.avx512f));
+    }
+
+    public SimdSupport getSimdSupport()
+    {
+        return simdSupport;
+    }
+}

core/trino-main/src/main/java/io/trino/testing/PlanTester.java

@@ -80,7 +80,6 @@
 import io.trino.memory.MemoryManagerConfig;
 import io.trino.memory.NodeMemoryConfig;
 import io.trino.metadata.AnalyzePropertyManager;
-import io.trino.metadata.BlockEncodingManager;
 import io.trino.metadata.CatalogManager;
 import io.trino.metadata.ColumnPropertyManager;
 import io.trino.metadata.DisabledSystemSecurityMetadata;
@@ -252,6 +251,7 @@
 import static io.trino.execution.ParameterExtractor.bindParameters;
 import static io.trino.execution.querystats.PlanOptimizersStatsCollector.createPlanOptimizersStatsCollector;
 import static io.trino.execution.warnings.WarningCollector.NOOP;
+import static io.trino.metadata.BlockEncodingManager.TESTING_BLOCK_ENCODING_MANAGER;
 import static io.trino.node.TestingInternalNodeManager.CURRENT_NODE;
 import static io.trino.spi.connector.Constraint.alwaysTrue;
 import static io.trino.spi.connector.DynamicFilter.EMPTY;
@@ -358,10 +358,9 @@ private PlanTester(Session defaultSession, int nodeCountForStats)
                 catalogManager,
                 notificationExecutor);
 
-        BlockEncodingManager blockEncodingManager = new BlockEncodingManager();
         TypeRegistry typeRegistry = new TypeRegistry(typeOperators, new FeaturesConfig());
         TypeManager typeManager = new InternalTypeManager(typeRegistry);
-        InternalBlockEncodingSerde blockEncodingSerde = new InternalBlockEncodingSerde(blockEncodingManager, typeManager);
+        InternalBlockEncodingSerde blockEncodingSerde = new InternalBlockEncodingSerde(TESTING_BLOCK_ENCODING_MANAGER, typeManager);
         SecretsResolver secretsResolver = new SecretsResolver(ImmutableMap.of());
 
         this.globalFunctionCatalog = new GlobalFunctionCatalog(
@@ -496,7 +495,7 @@ private PlanTester(Session defaultSession, int nodeCountForStats)
                 new GroupProviderManager(secretsResolver),
                 new SessionPropertyDefaults(nodeInfo, accessControl, secretsResolver),
                 typeRegistry,
-                blockEncodingManager,
+                TESTING_BLOCK_ENCODING_MANAGER,
                 new HandleResolver(),
                 exchangeManagerRegistry,
                 spoolingManagerRegistry);

core/trino-main/src/main/java/io/trino/util/MachineInfo.java

@@ -14,14 +14,25 @@
 package io.trino.util;
 
 import com.google.common.base.StandardSystemProperty;
+import com.google.common.collect.ImmutableSet;
 import oshi.SystemInfo;
+import oshi.hardware.CentralProcessor;
+import oshi.hardware.CentralProcessor.ProcessorIdentifier;
 
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static java.lang.Math.min;
+import static java.util.Locale.ENGLISH;
 
 public final class MachineInfo
 {
     // cache physical processor count, so that it's not queried multiple times during tests
     private static volatile int physicalProcessorCount = -1;
+    private static final SystemInfo SYSTEM_INFO = new SystemInfo();
 
     private MachineInfo() {}
 
@@ -38,7 +49,7 @@ public static int getAvailablePhysicalProcessorCount()
         if ("amd64".equals(osArch) || "x86_64".equals(osArch)) {
             // Oshi can recognize physical processor count (without hyper threading) for x86 platforms.
             // However, it doesn't correctly recognize physical processor count for ARM platforms.
-            totalPhysicalProcessorCount = new SystemInfo()
+            totalPhysicalProcessorCount = SYSTEM_INFO
                     .getHardware()
                     .getProcessor()
                     .getPhysicalProcessorCount();
@@ -52,4 +63,68 @@ public static int getAvailablePhysicalProcessorCount()
         physicalProcessorCount = min(totalPhysicalProcessorCount, availableProcessorCount);
         return physicalProcessorCount;
     }
+
+    public static ProcessorIdentifier getProcessorInfo()
+    {
+        return SYSTEM_INFO.getHardware().getProcessor().getProcessorIdentifier();
+    }
+
+    public static Set<String> readCpuFlags()
+    {
+        CentralProcessor cpu = SYSTEM_INFO.getHardware().getProcessor();
+        List<String> flags = cpu.getFeatureFlags();
+        if (flags == null || flags.isEmpty()) {
+            return ImmutableSet.of();
+        }
+        // Each element of flags represents the hardware support for an individual core, so we're want to calculate flags
+        // advertised by all cores
+        Set<String> intersection = null;
+
+        for (String line : flags) {
+            if (line == null || line.isBlank()) {
+                continue;
+            }
+
+            // Strip the "flags:" / "Features:" prefix if present.
+            String body = line;
+            int colon = line.indexOf(':');
+            if (colon >= 0) {
+                body = line.substring(colon + 1);
+            }
+
+            // Tokenize + normalize.
+            Set<String> tokens = Arrays.stream(body.trim().split("\\s+"))
+                    .map(token -> normalizeFlag(token))
+                    .filter(token -> !token.isEmpty())
+                    .collect(toImmutableSet());
+
+            if (tokens.isEmpty()) {
+                continue;
+            }
+
+            if (intersection == null) {
+                intersection = new HashSet<>(tokens);
+            }
+            else {
+                intersection.retainAll(tokens);
+                if (intersection.isEmpty()) {
+                    break; // nothing in common
+                }
+            }
+        }
+
+        return intersection == null ? ImmutableSet.of() : intersection;
+    }
+
+    public static String normalizeFlag(String flag)
+    {
+        flag = flag.toLowerCase(ENGLISH).replace("_", "").trim();
+
+        // Skip stray keys that may sneak in if the colon wasn’t found.
+        if (flag.equals("flags") || flag.equals("features")) {
+            return "";
+        }
+
+        return flag;
+    }
 }

core/trino-main/src/test/java/io/trino/execution/buffer/BenchmarkBlockSerde.java

@@ -81,7 +81,7 @@ public class BenchmarkBlockSerde
 {
     private static final DecimalType LONG_DECIMAL_TYPE = createDecimalType(30, 5);
 
-    public static final int ROWS = 10_000_000;
+    public static final int ROWS = 8192;
 
     @Benchmark
     public Object serializeLongDecimal(LongDecimalBenchmarkData data)