diff --git a/CMake/resolve_dependency_modules/inteliaa.cmake b/CMake/resolve_dependency_modules/inteliaa.cmake
new file mode 100644
index 000000000000..452fb4547ba0
--- /dev/null
+++ b/CMake/resolve_dependency_modules/inteliaa.cmake
@@ -0,0 +1,59 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# 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.
+include_guard(GLOBAL)
+
+set(VELOX_INTELIAA_VERSION 1.3.0)
+set(VELOX_INTELIAA_BUILD_SHA256_CHECKSUM
+    c3eba4d04a9d7aabcf26c9eaf81f6e9b26d19cb1b87a4a5f197a652cfa98f310)
+set(VELOX_INTELIAA_SOURCE_URL
+    "https://github.com/intel/qpl/archive/refs/tags/v${VELOX_INTELIAA_VERSION}.tar.gz"
+)
+
+resolve_dependency_url(INTELIAA)
+
+message(STATUS "Building Intel IAA from source")
+
+set(QPL_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/qpl_ep/install")
+set(QPL_STATIC_LIB_NAME
+    ${CMAKE_STATIC_LIBRARY_PREFIX}qpl${CMAKE_STATIC_LIBRARY_SUFFIX})
+set(QPL_STATIC_LIB "${QPL_PREFIX}/lib/${QPL_STATIC_LIB_NAME}")
+
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -ldl -laccel-config -L/usr/lib64")
+
+set(QPL_CMAKE_ARGS
+    -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
+    -DCMAKE_INSTALL_LIBDIR=${QPL_PREFIX}/lib
+    -DCMAKE_INSTALL_PREFIX=${QPL_PREFIX}
+    -DCMAKE_CXX_FLAGS=${CMAKE_CXX_FLAGS}
+    -DQPL_BUILD_TESTS=OFF
+    -DQPL_BUILD_EXAMPLES=OFF
+    -DQPL_LIB=ON)
+
+ExternalProject_Add(
+  intel_iaa
+  URL ${VELOX_INTELIAA_SOURCE_URL}
+  URL_HASH ${VELOX_INTELIAA_BUILD_SHA256_CHECKSUM}
+  BUILD_BYPRODUCTS "${QPL_STATIC_LIB}"
+  CMAKE_ARGS ${QPL_CMAKE_ARGS})
+
+file(MAKE_DIRECTORY "${QPL_PREFIX}/include")
+
+add_library(iaa::iaa UNKNOWN IMPORTED)
+set(QPL_LIBRARIES ${QPL_STATIC_LIB})
+set(QPL_INCLUDE_DIRS "${QPL_PREFIX}/include")
+set_target_properties(
+  iaa::iaa PROPERTIES IMPORTED_LOCATION ${QPL_LIBRARIES}
+                      INTERFACE_INCLUDE_DIRECTORIES ${QPL_INCLUDE_DIRS})
+
+add_dependencies(iaa::iaa intel_iaa-build)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index b9c88d4add33..8b1220546561 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -103,6 +103,7 @@ option(
   VELOX_ENABLE_INT64_BUILD_PARTITION_BOUND
   "make buildPartitionBounds_ a vector int64 instead of int32 to avoid integer overflow when the hashtable has billions of records"
   OFF)
+option(VELOX_ENABLE_INTEL_IAA "Enable Intel IAA support" OFF)
 
 # Explicitly force compilers to generate colored output. Compilers usually do
 # this by default if they detect the output is a terminal, but this assumption
@@ -256,6 +257,11 @@ if(VELOX_ENABLE_PARQUET)
   set(VELOX_ENABLE_ARROW ON)
 endif()
 
+if(VELOX_ENABLE_INTEL_IAA)
+  add_definitions(-DVELOX_ENABLE_INTEL_IAA)
+  message(STATUS "Intel IAA acceleration enabled")
+endif()
+
 # define processor variable for conditional compilation
 if(${VELOX_CODEGEN_SUPPORT})
   add_compile_definitions(CODEGEN_ENABLED=1)
@@ -474,6 +480,11 @@ if(NOT ${VELOX_BUILD_MINIMAL})
   include_directories(${Protobuf_INCLUDE_DIRS})
 endif()
 
+if(VELOX_ENABLE_INTEL_IAA)
+  set_source(inteliaa)
+  resolve_dependency(inteliaa)
+endif()
+
 # GCC needs to link a library to enable std::filesystem.
 if("${CMAKE_CXX_COMPILER_ID}" MATCHES "GNU")
   set(FILESYSTEM "stdc++fs")
diff --git a/velox/dwio/common/CMakeLists.txt b/velox/dwio/common/CMakeLists.txt
index 8334de75e0f5..c6562a4e2fcc 100644
--- a/velox/dwio/common/CMakeLists.txt
+++ b/velox/dwio/common/CMakeLists.txt
@@ -78,3 +78,9 @@ target_link_libraries(
   Boost::regex
   Folly::folly
   glog::glog)
+
+if(VELOX_ENABLE_INTEL_IAA)
+  add_library(velox_dwio_qpl QplJobPool.cpp)
+  target_link_libraries(velox_dwio_qpl iaa::iaa Folly::folly)
+  target_link_libraries(velox_dwio_common velox_dwio_qpl)
+endif()
diff --git a/velox/dwio/common/QplJobPool.cpp b/velox/dwio/common/QplJobPool.cpp
new file mode 100644
index 000000000000..bb5c7d108857
--- /dev/null
+++ b/velox/dwio/common/QplJobPool.cpp
@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#include "velox/dwio/common/QplJobPool.h"
+#include <folly/Random.h>
+#include "velox/common/base/Exceptions.h"
+
+namespace facebook::velox::dwio::common {
+
+std::array<std::atomic<bool>, QplJobHWPool::MAX_JOB_NUMBER>
+    QplJobHWPool::hwJobPtrLocks;
+
+QplJobHWPool& QplJobHWPool::getInstance() {
+  static QplJobHWPool pool;
+  return pool;
+}
+
+QplJobHWPool::QplJobHWPool() {
+  if (!iaaJobReady) {
+    allocateQPLJob();
+  }
+}
+
+QplJobHWPool::~QplJobHWPool() {
+  for (uint32_t i = 0; i < MAX_JOB_NUMBER; ++i) {
+    if (hwJobPtrPool[i]) {
+      qpl_fini_job(hwJobPtrPool[i]);
+      hwJobPtrPool[i] = nullptr;
+    }
+  }
+  iaaJobReady = false;
+}
+
+/**
+ * Allocate qpl job and put it into hwJobPtrPool
+ */
+void QplJobHWPool::allocateQPLJob() {
+  uint32_t job_size = 0;
+
+  // Get size required for saving a single qpl job object
+  qpl_get_job_size(qpl_path, &job_size);
+
+  // Allocate entire buffer for storing all job objects
+  hwJobsBuffer = std::make_unique<uint8_t[]>(job_size * MAX_JOB_NUMBER);
+
+  // Initialize pool for storing all job object pointers
+  // Allocate buffer by shifting address offset for each job object.
+  hwJobPtrPool.resize(MAX_JOB_NUMBER);
+  for (uint32_t i = 0; i < MAX_JOB_NUMBER; ++i) {
+    qpl_job* qplJobPtr =
+        reinterpret_cast<qpl_job*>(hwJobsBuffer.get() + i * job_size);
+    auto status = qpl_init_job(qpl_path, qplJobPtr);
+    if (status != QPL_STS_OK) {
+      iaaJobReady = false;
+      LOG(WARNING) << "Initialization of hardware IAA failed, statsu: "
+                   << status << ". Please check if Intel \
+        In-Memory Analytics Accelerator (IAA) is properly set up!";
+      return;
+    }
+    this->hwJobPtrPool[i] = qplJobPtr;
+    hwJobPtrLocks[i].store(false);
+  }
+
+  iaaJobReady = true;
+  return;
+}
+
+/**
+ * Acquire a deflate job.
+ * QplJobHWPool maintains MAX_JOB_NUMBER job slot to avoid frequently allocate,
+ * initialize and release job. Random slots is used to select a job and
+ * tryLockJob will check if the job is free.
+ * @return job_id and qpl_job pair
+ */
+std::pair<int, qpl_job*> QplJobHWPool::acquireDeflateJob() {
+  std::pair<int, qpl_job*> res;
+  res.first = -1;
+  res.second = nullptr;
+  if (!job_ready()) {
+    return res;
+  }
+  uint32_t retry = 0;
+  uint32_t index = folly::Random::rand32(1, MAX_JOB_NUMBER - 1);
+  while (!tryLockJob(index)) {
+    index = folly::Random::rand32(1, MAX_JOB_NUMBER - 1);
+    retry++;
+    if (retry > MAX_JOB_NUMBER) {
+      return res;
+    }
+  }
+  res.first = index;
+  if (index >= MAX_JOB_NUMBER) {
+    return res;
+  }
+  res.second = hwJobPtrPool[index];
+
+  return res;
+}
+
+void QplJobHWPool::releaseJob(int job_id) {
+  if (job_id >= MAX_JOB_NUMBER || job_id <= 0) {
+    return;
+  }
+  VELOX_CHECK_LT(job_id, MAX_JOB_NUMBER);
+  hwJobPtrLocks[job_id].store(false);
+  return;
+}
+
+bool QplJobHWPool::tryLockJob(uint32_t index) {
+  bool expected = false;
+  VELOX_CHECK_LT(index, MAX_JOB_NUMBER);
+  return hwJobPtrLocks[index].compare_exchange_strong(expected, true);
+}
+
+} // namespace facebook::velox::dwio::common
diff --git a/velox/dwio/common/QplJobPool.h b/velox/dwio/common/QplJobPool.h
new file mode 100644
index 000000000000..f80f3a02a704
--- /dev/null
+++ b/velox/dwio/common/QplJobPool.h
@@ -0,0 +1,97 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <mutex>
+#include <random>
+#include <vector>
+
+#include "qpl/qpl.h"
+
+namespace facebook::velox::dwio::common {
+
+// QplJobHWPool is resource pool to provide the job that will be submitted to
+// Intel® IAA Memory for Intel® IAA job will be allocated when the QPLJobHWPool
+// instance is created.
+//
+// The Intel® In-Memory Analytics Accelerator (Intel® IAA) is a hardware
+// accelerator that provides very high throughput compression and decompression
+// combined with primitive analytic functions. It primarily targets applications
+// such as big-data and in-memory analytic databases, as well as
+// application-transparent usages such as memory page compression. Intel® IAA
+// contains two main functional blocks: Compression and Analytics. The Analytics
+// pipe contains two sub-blocks: Decompress and Filter. These functions are tied
+// together, so that each analytics operation can perform decompress-only,
+// filter-only, or decompress-and-filter processing.
+//
+// Intel QPL is library to provide application programming interface (API) for
+// interaction with Intel® In-Memory Analytics Accelerator (Intel® IAA) hardware
+//
+// Intel® IAA:
+// https://www.intel.com/content/www/us/en/content-details/780887/intel-in-memory-analytics-accelerator-intel-iaa.html
+// Intel QPL:
+// https://intel.github.io/qpl/documentation/introduction_docs/introduction.html
+class QplJobHWPool {
+ public:
+  static QplJobHWPool& getInstance();
+  QplJobHWPool();
+  ~QplJobHWPool();
+
+  // Release QPL job by the job_id.
+  void releaseJob(int job_id);
+
+  // Return if the QPL job is allocated sucessfully.
+  const bool& job_ready() {
+    return iaaJobReady;
+  }
+
+  std::pair<int, qpl_job*> acquireDeflateJob();
+
+  /**
+   * Get qpl job by job id
+   * @param job_id the job id or index in the qpl job pool
+   * @return nullptr if the job id is invalid
+   */
+  qpl_job* getJobById(int job_id) {
+    if (job_id >= MAX_JOB_NUMBER || job_id <= 0) {
+      return nullptr;
+    }
+    return hwJobPtrPool[job_id];
+  }
+
+  static constexpr uint32_t MAX_JOB_NUMBER = 1024;
+
+ private:
+  bool tryLockJob(uint32_t index);
+  void allocateQPLJob();
+
+  qpl_path_t qpl_path = qpl_path_hardware;
+
+  // Max jobs in QPL_JOB_POOL
+  // Entire buffer for storing all job objects
+  std::unique_ptr<uint8_t[]> hwJobsBuffer;
+
+  // Job pool for storing all job object pointers
+  std::vector<qpl_job*> hwJobPtrPool;
+
+  // Locks for accessing each job object pointers
+  bool iaaJobReady;
+  static std::array<std::atomic<bool>, MAX_JOB_NUMBER> hwJobPtrLocks;
+};
+
+} // namespace facebook::velox::dwio::common
diff --git a/velox/dwio/common/compression/AsyncCompression.h b/velox/dwio/common/compression/AsyncCompression.h
new file mode 100644
index 000000000000..b6c01bdb8ab7
--- /dev/null
+++ b/velox/dwio/common/compression/AsyncCompression.h
@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#ifndef ASYNC_COMPRESSION_H_
+#define ASYNC_COMPRESSION_H_
+
+#include <folly/futures/Future.h>
+#include "velox/common/compression/Compression.h"
+
+namespace facebook::velox::dwio::common::compression {
+
+using facebook::velox::common::CompressionKind;
+
+class AsyncDecompressor {
+ public:
+  explicit AsyncDecompressor(){};
+
+  virtual ~AsyncDecompressor() = default;
+
+  virtual folly::SemiFuture<uint64_t> decompressAsync(
+      const char* src,
+      uint64_t srcLength,
+      char* dest,
+      uint64_t destLength) = 0;
+};
+
+std::unique_ptr<AsyncDecompressor> MakeIAAGzipCodec();
+
+/**
+ * Get the window size from zlib header(rfc1950).
+ * 0   1
+ * +---+---+
+ * |CMF|FLG|   (more-->)
+ * +---+---+
+ * bits 0 to 3  CM     Compression method
+ * bits 4 to 7  CINFO  Compression info
+ * CM (Compression method) This identifies the compression method used in the
+ * file. CM = 8 denotes the "deflate" compression method with a window size up
+ * to 32K. CINFO (Compression info) For CM = 8, CINFO is the base-2 logarithm of
+ * the LZ77 window size, minus eight (CINFO=7 indicates a 32K window size).
+ * @param stream_ptr the compressed block length for raw decompression
+ * @param stream_size compression options to use
+ */
+static int getZlibWindowBits(const uint8_t* stream_ptr, uint32_t stream_size) {
+  static constexpr uint8_t CM_ZLIB_DEFAULT_VALUE = 8u;
+  static constexpr uint32_t ZLIB_MIN_HEADER_SIZE = 2u;
+  static constexpr uint32_t ZLIB_INFO_OFFSET = 4u;
+  if (stream_size < ZLIB_MIN_HEADER_SIZE) {
+    return -1;
+  }
+  const uint8_t compression_method_and_flag = *stream_ptr++;
+  const uint8_t compression_method = compression_method_and_flag & 0xf;
+  const uint8_t compression_info =
+      compression_method_and_flag >> ZLIB_INFO_OFFSET;
+
+  if (CM_ZLIB_DEFAULT_VALUE != compression_method) {
+    return -1;
+  }
+  if (compression_info > 7) {
+    return -1;
+  }
+  return CM_ZLIB_DEFAULT_VALUE + compression_info;
+}
+
+/**
+ * Create a decompressor for the given compression kind in asynchronous mode.
+ * @param kind the compression type to implement
+ */
+static std::unique_ptr<dwio::common::compression::AsyncDecompressor>
+createAsyncDecompressor(facebook::velox::common::CompressionKind kind) {
+  switch (static_cast<int64_t>(kind)) {
+#ifdef VELOX_ENABLE_INTEL_IAA
+    case CompressionKind::CompressionKind_GZIP:
+      return MakeIAAGzipCodec();
+#endif
+    default:
+      LOG(WARNING) << "Asynchronous mode not support for compression codec  "
+                   << kind;
+      return nullptr;
+  }
+  return nullptr;
+}
+} // namespace facebook::velox::dwio::common::compression
+
+#endif
\ No newline at end of file
diff --git a/velox/dwio/common/compression/CMakeLists.txt b/velox/dwio/common/compression/CMakeLists.txt
index bda716ca7a25..579395281ae5 100644
--- a/velox/dwio/common/compression/CMakeLists.txt
+++ b/velox/dwio/common/compression/CMakeLists.txt
@@ -17,3 +17,11 @@ add_library(velox_dwio_common_compression Compression.cpp PagedInputStream.cpp
 
 target_link_libraries(velox_dwio_common_compression velox_dwio_common xsimd
                       Folly::folly)
+
+if(VELOX_ENABLE_INTEL_IAA)
+  add_library(velox_dwio_common_iaa_compression IAACompression.cpp)
+  target_link_libraries(velox_dwio_common_iaa_compression velox_dwio_qpl
+                        Folly::folly xsimd)
+  target_link_libraries(velox_dwio_common_compression
+                        velox_dwio_common_iaa_compression)
+endif()
diff --git a/velox/dwio/common/compression/IAACompression.cpp b/velox/dwio/common/compression/IAACompression.cpp
new file mode 100644
index 000000000000..139ec9494111
--- /dev/null
+++ b/velox/dwio/common/compression/IAACompression.cpp
@@ -0,0 +1,116 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#include "velox/common/base/Exceptions.h"
+#include "velox/dwio/common/QplJobPool.h"
+#include "velox/dwio/common/compression/AsyncCompression.h"
+
+namespace facebook::velox::dwio::common::compression {
+
+class GzipIAADecompressor : public AsyncDecompressor {
+ public:
+  explicit GzipIAADecompressor() {}
+
+  folly::SemiFuture<uint64_t> decompressAsync(
+      const char* src,
+      uint64_t srcLength,
+      char* dest,
+      uint64_t destLength) override;
+
+  int waitResult(int job_id);
+};
+
+folly::SemiFuture<uint64_t> GzipIAADecompressor::decompressAsync(
+    const char* src,
+    uint64_t srcLength,
+    char* dest,
+    uint64_t destLength) {
+  dwio::common::QplJobHWPool& qpl_job_pool =
+      dwio::common::QplJobHWPool::getInstance();
+  auto deflate_job = qpl_job_pool.acquireDeflateJob();
+  auto job = deflate_job.second;
+  if (job == nullptr) {
+    LOG(WARNING) << "cannot AcquireDeflateJob ";
+    return folly::makeSemiFutureWith([]() -> uint64_t {
+      throw std::runtime_error("Cannot acquire deflate job from pool");
+      return 0;
+    });
+  }
+  job->op = qpl_op_decompress;
+  job->next_in_ptr = reinterpret_cast<uint8_t*>(const_cast<char*>(src));
+  job->next_out_ptr = reinterpret_cast<uint8_t*>(dest);
+  job->available_in = static_cast<uint32_t>(srcLength);
+  job->available_out = static_cast<uint32_t>(destLength);
+  job->flags = QPL_FLAG_FIRST | QPL_FLAG_LAST | QPL_FLAG_ZLIB_MODE;
+
+  qpl_status status = qpl_submit_job(job);
+  if (status == QPL_STS_QUEUES_ARE_BUSY_ERR) {
+    qpl_job_pool.releaseJob(deflate_job.first);
+    deflate_job = qpl_job_pool.acquireDeflateJob();
+    job = deflate_job.second;
+    if (job == nullptr) {
+      LOG(WARNING)
+          << "cannot acqure deflate job after QPL_STS_QUEUES_ARE_BUSY_ERR ";
+      return folly::makeSemiFutureWith([]() -> uint64_t {
+        throw std::runtime_error(
+            "Cannot acqure deflate job from pool after QPL_STS_QUEUES_ARE_BUSY_ERR");
+        return 0;
+      });
+    }
+    job->op = qpl_op_decompress;
+    job->next_in_ptr = reinterpret_cast<uint8_t*>(const_cast<char*>(src));
+    job->next_out_ptr = reinterpret_cast<uint8_t*>(dest);
+    job->available_in = static_cast<uint32_t>(srcLength);
+    job->available_out = static_cast<uint32_t>(destLength);
+    job->flags = QPL_FLAG_FIRST | QPL_FLAG_LAST | QPL_FLAG_ZLIB_MODE;
+
+    status = qpl_submit_job(job);
+  }
+  if (status != QPL_STS_OK) {
+    qpl_job_pool.releaseJob(deflate_job.first);
+    LOG(WARNING) << "cannot submit job, error status: " << status;
+    return folly::makeSemiFutureWith([this, status]() -> uint64_t {
+      throw std::runtime_error("Cannot submit job, error status: " + status);
+      return 0;
+    });
+  } else {
+    return folly::makeSemiFuture().deferValue(
+        [this, deflate_job](auto&&) -> uint64_t {
+          return this->waitResult(deflate_job.first);
+        });
+  }
+}
+
+int GzipIAADecompressor::waitResult(int job_id) {
+  dwio::common::QplJobHWPool& qpl_job_pool =
+      dwio::common::QplJobHWPool::getInstance();
+  VELOX_CHECK_LT(job_id, qpl_job_pool.MAX_JOB_NUMBER);
+  qpl_job* job = qpl_job_pool.getJobById(job_id);
+
+  auto status = qpl_wait_job(job);
+  qpl_job_pool.releaseJob(job_id);
+  if (status == QPL_STS_OK) {
+    return 1;
+  }
+  LOG(WARNING) << "Decompress w/IAA error, status: " << status;
+  return 0;
+}
+
+std::unique_ptr<AsyncDecompressor> MakeIAAGzipCodec() {
+  return std::make_unique<GzipIAADecompressor>();
+}
+
+} // namespace facebook::velox::dwio::common::compression
\ No newline at end of file
diff --git a/velox/dwio/parquet/reader/CMakeLists.txt b/velox/dwio/parquet/reader/CMakeLists.txt
index 3fb5250b7e64..7254b9efec05 100644
--- a/velox/dwio/parquet/reader/CMakeLists.txt
+++ b/velox/dwio/parquet/reader/CMakeLists.txt
@@ -19,6 +19,7 @@ add_library(
   ParquetReader.cpp
   ParquetTypeWithId.cpp
   PageReader.cpp
+  IAAPageReader.cpp
   ParquetColumnReader.cpp
   ParquetData.cpp
   RepeatedColumnReader.cpp
diff --git a/velox/dwio/parquet/reader/IAAPageReader.cpp b/velox/dwio/parquet/reader/IAAPageReader.cpp
new file mode 100644
index 000000000000..07976158dd6e
--- /dev/null
+++ b/velox/dwio/parquet/reader/IAAPageReader.cpp
@@ -0,0 +1,507 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#include "velox/dwio/parquet/reader/IAAPageReader.h"
+#include "velox/dwio/common/BufferUtil.h"
+
+namespace facebook::velox::parquet {
+
+using thrift::Encoding;
+using thrift::PageHeader;
+
+void IAAPageReader::preDecompressPage(
+    bool& need_pre_decompress,
+    int64_t numValues) {
+  if (codec_ != common::CompressionKind::CompressionKind_GZIP) {
+    need_pre_decompress = false;
+    return;
+  }
+  for (;;) {
+    auto dataStart = pageStart_;
+    if (chunkSize_ <= pageStart_) {
+      // This may happen if seeking to exactly end of row group.
+      numRepDefsInPage_ = 0;
+      numRowsInPage_ = 0;
+      break;
+    }
+    PageHeader pageHeader = readPageHeader();
+    pageStart_ = pageDataStart_ + pageHeader.compressed_page_size;
+    switch (pageHeader.type) {
+      case thrift::PageType::DATA_PAGE:
+        prefetchDataPageV1(pageHeader);
+        break;
+      case thrift::PageType::DATA_PAGE_V2:
+        prefetchDataPageV2(pageHeader);
+        break;
+      case thrift::PageType::DICTIONARY_PAGE:
+        prefetchDictionary(pageHeader);
+        continue;
+      default:
+        break; // ignore INDEX page type and any other custom extensions
+    }
+    break;
+  }
+  need_pre_decompress = isWinSizeFit_;
+  rowGroupPageInfo_.numValues = numValues;
+  rowGroupPageInfo_.visitedRows = 0;
+}
+
+void IAAPageReader::prefetchNextPage() {
+  if (rowGroupPageInfo_.visitedRows + numRowsInPage_ >=
+      rowGroupPageInfo_.numValues) {
+    return;
+  }
+  if (chunkSize_ <= pageStart_) {
+    return;
+  }
+  PageHeader pageHeader = readPageHeader();
+  switch (pageHeader.type) {
+    case thrift::PageType::DATA_PAGE: {
+      dataPageHeader_ = pageHeader;
+      VELOX_CHECK(
+          pageHeader.type == thrift::PageType::DATA_PAGE &&
+          pageHeader.__isset.data_page_header);
+      rowGroupPageInfo_.dataPageData =
+          readBytes(pageHeader.compressed_page_size, pageBuffer_);
+      preDecompressData_ = iaaDecompress(
+          rowGroupPageInfo_.dataPageData,
+          pageHeader.compressed_page_size,
+          pageHeader.uncompressed_page_size,
+          rowGroupPageInfo_.uncompressedData,
+          dataDecompFuture);
+      break;
+    }
+    case thrift::PageType::DATA_PAGE_V2:
+      LOG(WARNING) << "Data Page V2 not support ";
+      break;
+    case thrift::PageType::DICTIONARY_PAGE:
+      LOG(WARNING) << "Wrong path ";
+      break;
+    default:
+      break; // ignore INDEX page type and any other custom extensions
+  }
+}
+
+bool IAAPageReader::seekToPreDecompPage(int64_t row) {
+  bool has_qpl = false;
+  if (this->dictDecompFuture.valid()) {
+    bool job_success = std::move(this->dictDecompFuture).get() > 0;
+    prepareDictionary(dictPageHeader_, job_success);
+    preDecompressDict_ = false;
+    has_qpl = true;
+  }
+
+  if (dataDecompFuture.valid()) {
+    bool job_success = std::move(this->dataDecompFuture).get() > 0;
+    prepareDataPageV1(dataPageHeader_, row, job_success);
+    preDecompressData_ = false;
+    has_qpl = true;
+  }
+
+  if (has_qpl) {
+    if (row == kRepDefOnly || row < rowOfPage_ + numRowsInPage_) {
+      return true;
+    }
+    updateRowInfoAfterPageSkipped();
+  }
+  return false;
+}
+
+void IAAPageReader::prefetchDataPageV1(const thrift::PageHeader& pageHeader) {
+  dataPageHeader_ = pageHeader;
+  VELOX_CHECK(
+      pageHeader.type == thrift::PageType::DATA_PAGE &&
+      pageHeader.__isset.data_page_header);
+
+  dataPageData_ = readBytes(pageHeader.compressed_page_size, pageBuffer_);
+  preDecompressData_ = iaaDecompress(
+      dataPageData_,
+      pageHeader.compressed_page_size,
+      pageHeader.uncompressed_page_size,
+      uncompressedDataV1Data_,
+      dataDecompFuture);
+  return;
+}
+
+void IAAPageReader::prefetchDataPageV2(const thrift::PageHeader& pageHeader) {
+  return;
+}
+
+void IAAPageReader::prefetchDictionary(const thrift::PageHeader& pageHeader) {
+  dictPageHeader_ = pageHeader;
+  dictionaryEncoding_ = pageHeader.dictionary_page_header.encoding;
+  VELOX_CHECK(
+      dictionaryEncoding_ == Encoding::PLAIN_DICTIONARY ||
+      dictionaryEncoding_ == Encoding::PLAIN);
+  dictPageData_ = readBytes(pageHeader.compressed_page_size, pageBuffer_);
+
+  preDecompressDict_ = iaaDecompress(
+      dictPageData_,
+      pageHeader.compressed_page_size,
+      pageHeader.uncompressed_page_size,
+      uncompressedDictData_,
+      dictDecompFuture);
+
+  return;
+}
+
+const bool IAAPageReader::iaaDecompress(
+    const char* pageData,
+    uint32_t compressedSize,
+    uint32_t uncompressedSize,
+    BufferPtr& uncompressedData,
+    folly::SemiFuture<uint64_t>& future) {
+  dwio::common::ensureCapacity<char>(
+      uncompressedData, uncompressedSize, &pool_);
+  static constexpr int PARQUET_ZLIB_WINDOW_BITS_4KB = 12;
+  future = folly::makeSemiFuture((uint64_t)0);
+  if (!isWinSizeFit_) {
+    // window size should be 4KB for IAA
+    if (PARQUET_ZLIB_WINDOW_BITS_4KB ==
+        dwio::common::compression::getZlibWindowBits(
+            (const uint8_t*)pageData, uncompressedSize)) {
+      isWinSizeFit_ = true;
+    } else {
+      future = folly::makeSemiFuture((uint64_t)0);
+      return true;
+    }
+  }
+  std::unique_ptr<dwio::common::compression::AsyncDecompressor> decompressor =
+      dwio::common::compression::createAsyncDecompressor(codec_);
+  if (decompressor == nullptr) {
+    return true;
+  }
+  auto decompFuture = decompressor->decompressAsync(
+      (const char*)pageData,
+      compressedSize,
+      (char*)uncompressedData->asMutable<char>(),
+      uncompressedSize);
+  if (decompFuture.isReady()) {
+    auto result = std::move(decompFuture).getTry();
+    if (result.hasException()) {
+      future = folly::makeSemiFuture((uint64_t)0);
+      return true;
+    }
+  }
+  future = std::move(decompFuture);
+  return true;
+}
+
+void IAAPageReader::seekToPage(int64_t row) {
+  this->defineDecoder_.reset();
+  this->repeatDecoder_.reset();
+  // 'rowOfPage_' is the row number of the first row of the next page.
+  this->rowOfPage_ += this->numRowsInPage_;
+
+  if (seekToPreDecompPage(row)) {
+    if (isWinSizeFit_) {
+      prefetchNextPage();
+    }
+    rowGroupPageInfo_.visitedRows += numRowsInPage_;
+    return;
+  }
+
+  for (;;) {
+    auto dataStart = pageStart_;
+    if (chunkSize_ <= pageStart_) {
+      // This may happen if seeking to exactly end of row group.
+      numRepDefsInPage_ = 0;
+      numRowsInPage_ = 0;
+      break;
+    }
+    PageHeader pageHeader = this->readPageHeader();
+    pageStart_ = pageDataStart_ + pageHeader.compressed_page_size;
+
+    switch (pageHeader.type) {
+      case thrift::PageType::DATA_PAGE:
+        prepareDataPageV1(pageHeader, row);
+        break;
+      case thrift::PageType::DATA_PAGE_V2:
+        prepareDataPageV2(pageHeader, row);
+        break;
+      case thrift::PageType::DICTIONARY_PAGE:
+        if (row == kRepDefOnly) {
+          skipBytes(
+              pageHeader.compressed_page_size,
+              inputStream_.get(),
+              bufferStart_,
+              bufferEnd_);
+          continue;
+        }
+        prepareDictionary(pageHeader);
+        continue;
+      default:
+        break; // ignore INDEX page type and any other custom extensions
+    }
+    if (row == kRepDefOnly || row < rowOfPage_ + numRowsInPage_) {
+      break;
+    }
+    this->updateRowInfoAfterPageSkipped();
+  }
+  if (isWinSizeFit_) {
+    prefetchNextPage();
+  }
+  rowGroupPageInfo_.visitedRows += numRowsInPage_;
+}
+
+void IAAPageReader::prepareDataPageV1(
+    const PageHeader& pageHeader,
+    int64_t row,
+    bool job_success) {
+  VELOX_CHECK(
+      pageHeader.type == thrift::PageType::DATA_PAGE &&
+      pageHeader.__isset.data_page_header);
+  numRepDefsInPage_ = pageHeader.data_page_header.num_values;
+  setPageRowInfo(row == kRepDefOnly);
+  if (row != kRepDefOnly && numRowsInPage_ != kRowsUnknown &&
+      numRowsInPage_ + rowOfPage_ <= row) {
+    dwio::common::skipBytes(
+        pageHeader.compressed_page_size,
+        inputStream_.get(),
+        bufferStart_,
+        bufferEnd_);
+
+    return;
+  }
+  if (job_success) {
+    if (rowGroupPageInfo_.visitedRows > 0) {
+      BufferPtr tmp = uncompressedDataV1Data_;
+      uncompressedDataV1Data_ = rowGroupPageInfo_.uncompressedData;
+      rowGroupPageInfo_.uncompressedData = tmp;
+    }
+    pageData_ = uncompressedDataV1Data_->as<char>();
+  } else {
+    if (!preDecompressData_) {
+      dataPageData_ = readBytes(pageHeader.compressed_page_size, pageBuffer_);
+    } else if (rowGroupPageInfo_.visitedRows > 0) {
+      dataPageData_ = rowGroupPageInfo_.dataPageData;
+    }
+    pageData_ = decompressData(
+        dataPageData_,
+        pageHeader.compressed_page_size,
+        pageHeader.uncompressed_page_size);
+  }
+  auto pageEnd = pageData_ + pageHeader.uncompressed_page_size;
+  if (maxRepeat_ > 0) {
+    uint32_t repeatLength = readField<int32_t>(pageData_);
+    repeatDecoder_ = std::make_unique<::arrow::util::RleDecoder>(
+        reinterpret_cast<const uint8_t*>(pageData_),
+        repeatLength,
+        ::arrow::bit_util::NumRequiredBits(maxRepeat_));
+
+    pageData_ += repeatLength;
+  }
+
+  if (maxDefine_ > 0) {
+    auto defineLength = readField<uint32_t>(pageData_);
+    if (maxDefine_ == 1) {
+      defineDecoder_ = std::make_unique<RleBpDecoder>(
+          pageData_,
+          pageData_ + defineLength,
+          ::arrow::bit_util::NumRequiredBits(maxDefine_));
+    }
+    wideDefineDecoder_ = std::make_unique<::arrow::util::RleDecoder>(
+        reinterpret_cast<const uint8_t*>(pageData_),
+        defineLength,
+        ::arrow::bit_util::NumRequiredBits(maxDefine_));
+    pageData_ += defineLength;
+  }
+  encodedDataSize_ = pageEnd - pageData_;
+
+  encoding_ = pageHeader.data_page_header.encoding;
+  if (!hasChunkRepDefs_ && (numRowsInPage_ == kRowsUnknown || maxDefine_ > 1)) {
+    readPageDefLevels();
+  }
+
+  if (row != kRepDefOnly) {
+    makeDecoder();
+  }
+}
+
+void IAAPageReader::prepareDictionary(
+    const PageHeader& pageHeader,
+    bool job_success) {
+  dictionary_.numValues = pageHeader.dictionary_page_header.num_values;
+  dictionaryEncoding_ = pageHeader.dictionary_page_header.encoding;
+  dictionary_.sorted = pageHeader.dictionary_page_header.__isset.is_sorted &&
+      pageHeader.dictionary_page_header.is_sorted;
+  VELOX_CHECK(
+      dictionaryEncoding_ == Encoding::PLAIN_DICTIONARY ||
+      dictionaryEncoding_ == Encoding::PLAIN);
+
+  if (codec_ != common::CompressionKind::CompressionKind_NONE) {
+    if (job_success) {
+      pageData_ = uncompressedDictData_->as<char>();
+    } else {
+      if (!preDecompressDict_) {
+        dictPageData_ = readBytes(pageHeader.compressed_page_size, pageBuffer_);
+      }
+      pageData_ = decompressData(
+          dictPageData_,
+          pageHeader.compressed_page_size,
+          pageHeader.uncompressed_page_size);
+    }
+  }
+
+  auto parquetType = type_->parquetType_.value();
+  switch (parquetType) {
+    case thrift::Type::INT32:
+    case thrift::Type::INT64:
+    case thrift::Type::FLOAT:
+    case thrift::Type::DOUBLE: {
+      int32_t typeSize = (parquetType == thrift::Type::INT32 ||
+                          parquetType == thrift::Type::FLOAT)
+          ? sizeof(float)
+          : sizeof(double);
+      auto numBytes = dictionary_.numValues * typeSize;
+      if (type_->type()->isShortDecimal() &&
+          parquetType == thrift::Type::INT32) {
+        auto veloxTypeLength = type_->type()->cppSizeInBytes();
+        auto numVeloxBytes = dictionary_.numValues * veloxTypeLength;
+        dictionary_.values =
+            AlignedBuffer::allocate<char>(numVeloxBytes, &pool_);
+      } else {
+        dictionary_.values = AlignedBuffer::allocate<char>(numBytes, &pool_);
+      }
+      if (pageData_) {
+        memcpy(dictionary_.values->asMutable<char>(), pageData_, numBytes);
+      } else {
+        dwio::common::readBytes(
+            numBytes,
+            inputStream_.get(),
+            dictionary_.values->asMutable<char>(),
+            bufferStart_,
+            bufferEnd_);
+      }
+      if (type_->type()->isShortDecimal() &&
+          parquetType == thrift::Type::INT32) {
+        auto values = dictionary_.values->asMutable<int64_t>();
+        auto parquetValues = dictionary_.values->asMutable<int32_t>();
+        for (auto i = dictionary_.numValues - 1; i >= 0; --i) {
+          // Expand the Parquet type length values to Velox type length.
+          // We start from the end to allow in-place expansion.
+          values[i] = parquetValues[i];
+        }
+      }
+      break;
+    }
+    case thrift::Type::BYTE_ARRAY: {
+      dictionary_.values =
+          AlignedBuffer::allocate<StringView>(dictionary_.numValues, &pool_);
+      auto numBytes = pageHeader.uncompressed_page_size;
+      auto values = dictionary_.values->asMutable<StringView>();
+      dictionary_.strings = AlignedBuffer::allocate<char>(numBytes, &pool_);
+      auto strings = dictionary_.strings->asMutable<char>();
+      if (pageData_) {
+        memcpy(strings, pageData_, numBytes);
+      } else {
+        dwio::common::readBytes(
+            numBytes, inputStream_.get(), strings, bufferStart_, bufferEnd_);
+      }
+      auto header = strings;
+      for (auto i = 0; i < dictionary_.numValues; ++i) {
+        auto length = *reinterpret_cast<const int32_t*>(header);
+        values[i] = StringView(header + sizeof(int32_t), length);
+        header += length + sizeof(int32_t);
+      }
+      VELOX_CHECK_EQ(header, strings + numBytes);
+      break;
+    }
+    case thrift::Type::FIXED_LEN_BYTE_ARRAY: {
+      auto parquetTypeLength = type_->typeLength_;
+      auto numParquetBytes = dictionary_.numValues * parquetTypeLength;
+      auto veloxTypeLength = type_->type()->cppSizeInBytes();
+      auto numVeloxBytes = dictionary_.numValues * veloxTypeLength;
+      dictionary_.values = AlignedBuffer::allocate<char>(numVeloxBytes, &pool_);
+      auto data = dictionary_.values->asMutable<char>();
+      // Read the data bytes.
+      if (pageData_) {
+        memcpy(data, pageData_, numParquetBytes);
+      } else {
+        dwio::common::readBytes(
+            numParquetBytes,
+            inputStream_.get(),
+            data,
+            bufferStart_,
+            bufferEnd_);
+      }
+      if (type_->type()->isShortDecimal()) {
+        // Parquet decimal values have a fixed typeLength_ and are in big-endian
+        // layout.
+        if (numParquetBytes < numVeloxBytes) {
+          auto values = dictionary_.values->asMutable<int64_t>();
+          for (auto i = dictionary_.numValues - 1; i >= 0; --i) {
+            // Expand the Parquet type length values to Velox type length.
+            // We start from the end to allow in-place expansion.
+            auto sourceValue = data + (i * parquetTypeLength);
+            int64_t value = *sourceValue >= 0 ? 0 : -1;
+            memcpy(
+                reinterpret_cast<uint8_t*>(&value) + veloxTypeLength -
+                    parquetTypeLength,
+                sourceValue,
+                parquetTypeLength);
+            values[i] = value;
+          }
+        }
+        auto values = dictionary_.values->asMutable<int64_t>();
+        for (auto i = 0; i < dictionary_.numValues; ++i) {
+          values[i] = __builtin_bswap64(values[i]);
+        }
+        break;
+      } else if (type_->type()->isLongDecimal()) {
+        // Parquet decimal values have a fixed typeLength_ and are in big-endian
+        // layout.
+        if (numParquetBytes < numVeloxBytes) {
+          auto values = dictionary_.values->asMutable<int128_t>();
+          for (auto i = dictionary_.numValues - 1; i >= 0; --i) {
+            // Expand the Parquet type length values to Velox type length.
+            // We start from the end to allow in-place expansion.
+            auto sourceValue = data + (i * parquetTypeLength);
+            int128_t value = *sourceValue >= 0 ? 0 : -1;
+            memcpy(
+                reinterpret_cast<uint8_t*>(&value) + veloxTypeLength -
+                    parquetTypeLength,
+                sourceValue,
+                parquetTypeLength);
+            values[i] = value;
+          }
+        }
+        auto values = dictionary_.values->asMutable<int128_t>();
+        for (auto i = 0; i < dictionary_.numValues; ++i) {
+          values[i] = bits::builtin_bswap128(values[i]);
+        }
+        break;
+      }
+      VELOX_UNSUPPORTED(
+          "Parquet type {} not supported for dictionary", parquetType);
+    }
+    case thrift::Type::INT96:
+    default:
+      VELOX_UNSUPPORTED(
+          "Parquet type {} not supported for dictionary", parquetType);
+  }
+}
+
+IAAPageReader::~IAAPageReader() {
+  if (dataDecompFuture.valid()) {
+    std::move(dataDecompFuture).get();
+  }
+
+  if (dictDecompFuture.valid()) {
+    std::move(dictDecompFuture).get();
+  }
+}
+} // namespace facebook::velox::parquet
\ No newline at end of file
diff --git a/velox/dwio/parquet/reader/IAAPageReader.h b/velox/dwio/parquet/reader/IAAPageReader.h
new file mode 100644
index 000000000000..dc331465b798
--- /dev/null
+++ b/velox/dwio/parquet/reader/IAAPageReader.h
@@ -0,0 +1,114 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include <folly/futures/Future.h>
+
+#include "velox/dwio/common/compression/AsyncCompression.h"
+#include "velox/dwio/parquet/reader/PageReader.h"
+
+namespace facebook::velox::parquet {
+
+using folly::SemiFuture;
+
+struct PreDecompPageInfo {
+  int64_t numValues; // Number of values in this row group
+  int64_t visitedRows; // rows already read
+  const char* FOLLY_NULLABLE dataPageData{nullptr};
+  BufferPtr uncompressedData;
+};
+
+class IAAPageReader : public PageReader {
+ public:
+  IAAPageReader(
+      std::unique_ptr<dwio::common::SeekableInputStream> stream,
+      memory::MemoryPool& pool,
+      ParquetTypeWithIdPtr fileType,
+      common::CompressionKind codec,
+      int64_t chunkSize)
+      : PageReader(std::move(stream), pool, fileType, codec, chunkSize) {
+    uncompressedDictData_ = nullptr;
+    uncompressedDataV1Data_ = nullptr;
+  }
+  ~IAAPageReader();
+
+  PageReaderType getType() {
+    return PageReaderType::IAA;
+  };
+
+  /**
+   * Submit decompression job to IAA, store the decompression future
+   * @param need_pre_decompress true if the codec and window bits are adaptable
+   * for IAA
+   * @param numValues number of values in row group. This value is stored in
+   * rowGroupPageInfo_ and used for determine whether there is need to
+   * pre-decompress by IAA
+   */
+  void preDecompressPage(bool& need_pre_decompress, int64_t numValues);
+
+  // Override method to call seekToPreDecompPage and
+  // prefetchNextPage in IAAPageReader
+  virtual void seekToPage(int64_t row);
+
+ private:
+  void prefetchDataPageV1(const thrift::PageHeader& pageHeader);
+  void prefetchDataPageV2(const thrift::PageHeader& pageHeader);
+  void prefetchDictionary(const thrift::PageHeader& pageHeader);
+
+  // Prefetch the next page if there are more than one page in the row group
+  // and then submit the page decompression job to IAA
+  void prefetchNextPage();
+
+  // Get decompressed page from IAA async decompressor. Then Reads and sets
+  // 'rowOfPage_' and 'numRowsInPage_' and initializes a decoder for the found
+  // page.
+  bool seekToPreDecompPage(int64_t row);
+
+  void prepareDataPageV1(
+      const thrift::PageHeader& pageHeader,
+      int64_t row,
+      bool job_success = false);
+  void prepareDictionary(
+      const thrift::PageHeader& pageHeader,
+      bool job_success = false);
+  const bool iaaDecompress(
+      const char* FOLLY_NONNULL pageData,
+      uint32_t compressedSize,
+      uint32_t uncompressedSize,
+      BufferPtr& uncompressedData,
+      SemiFuture<uint64_t>& future);
+
+  // Used for pre-decompress
+  BufferPtr uncompressedDictData_;
+  BufferPtr uncompressedDataV1Data_;
+  thrift::PageHeader dictPageHeader_;
+  const char* FOLLY_NULLABLE dictPageData_{nullptr};
+  bool needUncompressDict;
+
+  thrift::PageHeader dataPageHeader_;
+  const char* FOLLY_NULLABLE dataPageData_{nullptr};
+
+  SemiFuture<uint64_t> dictDecompFuture = SemiFuture<uint64_t>::makeEmpty();
+  SemiFuture<uint64_t> dataDecompFuture = SemiFuture<uint64_t>::makeEmpty();
+
+  bool preDecompressDict_ = false;
+  bool preDecompressData_ = false;
+  bool isWinSizeFit_ = false;
+  PreDecompPageInfo rowGroupPageInfo_;
+};
+
+} // namespace facebook::velox::parquet
diff --git a/velox/dwio/parquet/reader/PageReader.h b/velox/dwio/parquet/reader/PageReader.h
index 4ef9c959b0db..ef75075bde2f 100644
--- a/velox/dwio/parquet/reader/PageReader.h
+++ b/velox/dwio/parquet/reader/PageReader.h
@@ -23,6 +23,7 @@
 #include "velox/dwio/common/compression/Compression.h"
 #include "velox/dwio/parquet/reader/BooleanDecoder.h"
 #include "velox/dwio/parquet/reader/DeltaBpDecoder.h"
+#include "velox/dwio/parquet/reader/PageReaderBase.h"
 #include "velox/dwio/parquet/reader/ParquetTypeWithId.h"
 #include "velox/dwio/parquet/reader/RleBpDataDecoder.h"
 #include "velox/dwio/parquet/reader/StringDecoder.h"
@@ -34,7 +35,7 @@ namespace facebook::velox::parquet {
 /// Manages access to pages inside a ColumnChunk. Interprets page headers and
 /// encodings and presents the combination of pages and encoded values as a
 /// continuous stream accessible via readWithVisitor().
-class PageReader {
+class PageReader : public PageReaderBase {
  public:
   PageReader(
       std::unique_ptr<dwio::common::SeekableInputStream> stream,
@@ -69,6 +70,10 @@ class PageReader {
         chunkSize_(chunkSize),
         nullConcatenation_(pool_) {}
 
+  PageReaderType getType() {
+    return PageReaderType::Common;
+  };
+
   /// Advances 'numRows' top level rows.
   void skip(int64_t numRows);
 
@@ -137,7 +142,7 @@ class PageReader {
   // bufferEnd_ to the corresponding positions.
   thrift::PageHeader readPageHeader();
 
- private:
+ protected:
   // Indicates that we only want the repdefs for the next page. Used when
   // prereading repdefs with seekToPage.
   static constexpr int64_t kRepDefOnly = -1;
@@ -169,7 +174,7 @@ class PageReader {
   // is interpreted in terms of leaf rows, including leaf
   // nulls. Seeking ahead of pages covered by decodeRepDefs is not
   // allowed for non-top level columns.
-  void seekToPage(int64_t row);
+  virtual void seekToPage(int64_t row);
 
   // Preloads the repdefs for the column chunk. To avoid preloading,
   // would need a way too clone the input stream so that one stream
diff --git a/velox/dwio/parquet/reader/PageReaderBase.h b/velox/dwio/parquet/reader/PageReaderBase.h
new file mode 100644
index 000000000000..51af5b8b7cca
--- /dev/null
+++ b/velox/dwio/parquet/reader/PageReaderBase.h
@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#pragma once
+
+namespace facebook::velox::parquet {
+
+enum PageReaderType {
+  Common = 0,
+  IAA = 1,
+};
+
+class PageReaderBase {
+ public:
+  //   explicit PageReaderBase(){};
+
+  virtual ~PageReaderBase(){};
+
+  virtual PageReaderType getType() = 0;
+
+  /**
+   * skips 'numValues' top level rows, touching null flags only.
+   * Non-null values are not prepared for reading.
+   * @param numValues
+   * @return void
+   */
+  virtual void skipNullsOnly(int64_t numValues) = 0;
+
+  /**
+   * Reads 'numValues' null flags into 'nulls' and advances the
+   * decoders by as much. The read may span several pages. If there
+   * are no nulls, buffer may be set to nullptr.
+   * @param numValues
+   * @param buffer
+   * @return void
+   */
+  virtual void readNullsOnly(int64_t numValues, BufferPtr& buffer) = 0;
+
+  /**
+   * Advances 'numRows' top level rows.
+   * @param numRows
+   * @return void
+   */
+  virtual void skip(int64_t numRows) = 0;
+
+  /* Returns the current string dictionary as a FlatVector<StringView>.
+   * @param type
+   * @return VectorPtr
+   */
+  virtual const VectorPtr& dictionaryValues(const TypePtr& type) = 0;
+
+  virtual bool isDictionary() const = 0;
+
+  virtual void clearDictionary() = 0;
+
+  virtual bool isDeltaBinaryPacked() const = 0;
+};
+
+} // namespace facebook::velox::parquet
\ No newline at end of file
diff --git a/velox/dwio/parquet/reader/ParquetData.cpp b/velox/dwio/parquet/reader/ParquetData.cpp
index 283190bbfb0a..0c95f065e01c 100644
--- a/velox/dwio/parquet/reader/ParquetData.cpp
+++ b/velox/dwio/parquet/reader/ParquetData.cpp
@@ -81,6 +81,30 @@ bool ParquetData::rowGroupMatches(
   return true;
 }
 
+bool ParquetData::preDecompRowGroup(uint32_t index) {
+#ifndef VELOX_ENABLE_INTEL_IAA
+  return false;
+#endif
+  auto metaData = fileMetaDataPtr_.rowGroup(index).columnChunk(type_->column());
+  if (metaData.compression() != common::CompressionKind::CompressionKind_GZIP ||
+      !needPreDecomp) {
+    LOG(WARNING) << "QPL Job not ready or zlib window size(" << needPreDecomp
+                 << ") is not 4KB";
+    return false;
+  }
+
+  pageReaders_.resize(fileMetaDataPtr_.numRowGroups());
+  auto iaaPageReader = std::make_unique<IAAPageReader>(
+      std::move(streams_[index]),
+      pool_,
+      type_,
+      metaData.compression(),
+      metaData.totalCompressedSize());
+  iaaPageReader->preDecompressPage(needPreDecomp, metaData.numValues());
+  pageReaders_[index] = std::move(iaaPageReader);
+  return needPreDecomp;
+}
+
 void ParquetData::enqueueRowGroup(
     uint32_t index,
     dwio::common::BufferedInput& input) {
@@ -113,6 +137,11 @@ dwio::common::PositionProvider ParquetData::seekToRowGroup(uint32_t index) {
   VELOX_CHECK_LT(index, streams_.size());
   VELOX_CHECK(streams_[index], "Stream not enqueued for column");
   auto metadata = fileMetaDataPtr_.rowGroup(index).columnChunk(type_->column());
+  if (metadata.compression() == common::CompressionKind::CompressionKind_GZIP &&
+      pageReaders_.size() > index && pageReaders_[index] != nullptr) {
+    reader_ = std::move(pageReaders_[index]);
+    return dwio::common::PositionProvider(empty);
+  }
   reader_ = std::make_unique<PageReader>(
       std::move(streams_[index]),
       pool_,
diff --git a/velox/dwio/parquet/reader/ParquetData.h b/velox/dwio/parquet/reader/ParquetData.h
index e75e69558b8e..bfc71d7e42cd 100644
--- a/velox/dwio/parquet/reader/ParquetData.h
+++ b/velox/dwio/parquet/reader/ParquetData.h
@@ -17,6 +17,7 @@
 #pragma once
 
 #include "velox/dwio/common/BufferUtil.h"
+#include "velox/dwio/parquet/reader/IAAPageReader.h"
 #include "velox/dwio/parquet/reader/Metadata.h"
 #include "velox/dwio/parquet/reader/PageReader.h"
 
@@ -62,6 +63,9 @@ class ParquetData : public dwio::common::FormatData {
   /// Prepares to read data for 'index'th row group.
   void enqueueRowGroup(uint32_t index, dwio::common::BufferedInput& input);
 
+  /// pre-decompress data for the 'index'th row group
+  bool preDecompRowGroup(uint32_t index);
+
   /// Positions 'this' at 'index'th row group. loadRowGroup must be called
   /// first. The returned PositionProvider is empty and should not be used.
   /// Other formats may use it.
@@ -74,7 +78,7 @@ class ParquetData : public dwio::common::FormatData {
       FilterRowGroupsResult&) override;
 
   PageReader* FOLLY_NONNULL reader() const {
-    return reader_.get();
+    return dynamic_cast<PageReader*>(reader_.get());
   }
 
   // Reads null flags for 'numValues' next top level rows. The first 'numValues'
@@ -160,7 +164,11 @@ class ParquetData : public dwio::common::FormatData {
   /// PageReader::readWithVisitor().
   template <typename Visitor>
   void readWithVisitor(Visitor visitor) {
-    reader_->readWithVisitor(visitor);
+    if (reader_->getType() == PageReaderType::IAA) {
+      dynamic_cast<IAAPageReader*>(reader_.get())->readWithVisitor(visitor);
+    } else {
+      dynamic_cast<PageReader*>(reader_.get())->readWithVisitor(visitor);
+    }
   }
 
   const VectorPtr& dictionaryValues(const TypePtr& type) {
@@ -202,8 +210,9 @@ class ParquetData : public dwio::common::FormatData {
   const uint32_t maxDefine_;
   const uint32_t maxRepeat_;
   int64_t rowsInRowGroup_;
-  std::unique_ptr<PageReader> reader_;
-
+  std::unique_ptr<PageReaderBase> reader_;
+  std::vector<std::unique_ptr<PageReaderBase>> pageReaders_;
+  bool needPreDecomp = true;
   // Nulls derived from leaf repdefs for non-leaf readers.
   BufferPtr presetNulls_;
 
diff --git a/velox/dwio/parquet/reader/ParquetReader.cpp b/velox/dwio/parquet/reader/ParquetReader.cpp
index 7d9a7d97da19..b0f286123256 100644
--- a/velox/dwio/parquet/reader/ParquetReader.cpp
+++ b/velox/dwio/parquet/reader/ParquetReader.cpp
@@ -117,6 +117,7 @@ class ReaderBase {
   std::shared_ptr<const dwio::common::TypeWithId> schemaWithId_;
 
   const bool binaryAsString = false;
+  bool needPreDecomp = true;
 
   // Map from row group index to pre-created loading BufferedInput.
   std::unordered_map<uint32_t, std::shared_ptr<dwio::common::BufferedInput>>
@@ -137,6 +138,7 @@ ReaderBase::ReaderBase(
 
   loadFileMetaData();
   initializeSchema();
+  needPreDecomp = true;
 }
 
 void ReaderBase::loadFileMetaData() {
@@ -600,6 +602,9 @@ void ReaderBase::scheduleRowGroups(
     auto thisGroup = rowGroupIds[currentGroup + i];
     if (!inputs_[thisGroup]) {
       inputs_[thisGroup] = reader.loadRowGroup(thisGroup, input_);
+      if (needPreDecomp) {
+        needPreDecomp = reader.preDecompRowGroup(thisGroup);
+      }
     }
   }
 
diff --git a/velox/dwio/parquet/reader/RepeatedColumnReader.cpp b/velox/dwio/parquet/reader/RepeatedColumnReader.cpp
index 250bd204e083..afb68ecd0a01 100644
--- a/velox/dwio/parquet/reader/RepeatedColumnReader.cpp
+++ b/velox/dwio/parquet/reader/RepeatedColumnReader.cpp
@@ -83,6 +83,21 @@ void skipUnreadLengthsAndNulls(dwio::common::SelectiveColumnReader& reader) {
   }
 }
 
+void preDecompChild(
+    dwio::common::SelectiveColumnReader* reader,
+    uint32_t index) {
+#ifdef VELOX_ENABLE_INTEL_IAA
+  auto children = reader->children();
+  if (children.empty()) {
+    reader->formatData().as<ParquetData>().preDecompRowGroup(index);
+    return;
+  }
+  for (auto* child : children) {
+    preDecompChild(child, index);
+  }
+#endif
+}
+
 void enqueueChildren(
     dwio::common::SelectiveColumnReader* reader,
     uint32_t index,
@@ -137,6 +152,10 @@ void MapColumnReader::enqueueRowGroup(
   enqueueChildren(this, index, input);
 }
 
+void MapColumnReader::preDecompRowGroup(uint32_t index) {
+  preDecompChild(this, index);
+}
+
 void MapColumnReader::seekToRowGroup(uint32_t index) {
   SelectiveMapColumnReader::seekToRowGroup(index);
   readOffset_ = 0;
@@ -243,6 +262,10 @@ void ListColumnReader::enqueueRowGroup(
   enqueueChildren(this, index, input);
 }
 
+void ListColumnReader::preDecompRowGroup(uint32_t index) {
+  preDecompChild(this, index);
+}
+
 void ListColumnReader::seekToRowGroup(uint32_t index) {
   SelectiveListColumnReader::seekToRowGroup(index);
   readOffset_ = 0;
diff --git a/velox/dwio/parquet/reader/RepeatedColumnReader.h b/velox/dwio/parquet/reader/RepeatedColumnReader.h
index 3155e8d66478..7d9c831ff1c5 100644
--- a/velox/dwio/parquet/reader/RepeatedColumnReader.h
+++ b/velox/dwio/parquet/reader/RepeatedColumnReader.h
@@ -71,6 +71,7 @@ class MapColumnReader : public dwio::common::SelectiveMapColumnReader {
   void seekToRowGroup(uint32_t index) override;
 
   void enqueueRowGroup(uint32_t index, dwio::common::BufferedInput& input);
+  void preDecompRowGroup(uint32_t index);
 
   void read(
       vector_size_t offset,
@@ -128,6 +129,7 @@ class ListColumnReader : public dwio::common::SelectiveListColumnReader {
 
   void enqueueRowGroup(uint32_t index, dwio::common::BufferedInput& input);
 
+  void preDecompRowGroup(uint32_t index);
   void read(
       vector_size_t offset,
       RowSet rows,
diff --git a/velox/dwio/parquet/reader/StructColumnReader.cpp b/velox/dwio/parquet/reader/StructColumnReader.cpp
index eca887eab155..bcbc65f82166 100644
--- a/velox/dwio/parquet/reader/StructColumnReader.cpp
+++ b/velox/dwio/parquet/reader/StructColumnReader.cpp
@@ -140,6 +140,25 @@ void StructColumnReader::enqueueRowGroup(
   }
 }
 
+bool StructColumnReader::preDecompRowGroup(uint32_t index) {
+  for (auto& child : children_) {
+    if (!needPreDecomp) {
+      return false;
+    }
+    if (auto structChild = dynamic_cast<StructColumnReader*>(child)) {
+      structChild->preDecompRowGroup(index);
+    } else if (auto listChild = dynamic_cast<ListColumnReader*>(child)) {
+      listChild->preDecompRowGroup(index);
+    } else if (auto mapChild = dynamic_cast<MapColumnReader*>(child)) {
+      mapChild->preDecompRowGroup(index);
+    } else {
+      needPreDecomp =
+          child->formatData().as<ParquetData>().preDecompRowGroup(index);
+    }
+  }
+  return needPreDecomp;
+}
+
 void StructColumnReader::seekToRowGroup(uint32_t index) {
   SelectiveStructColumnReader::seekToRowGroup(index);
   BufferPtr noBuffer;
diff --git a/velox/dwio/parquet/reader/StructColumnReader.h b/velox/dwio/parquet/reader/StructColumnReader.h
index f38c9e849c73..e9485554c7f0 100644
--- a/velox/dwio/parquet/reader/StructColumnReader.h
+++ b/velox/dwio/parquet/reader/StructColumnReader.h
@@ -49,6 +49,8 @@ class StructColumnReader : public dwio::common::SelectiveStructColumnReader {
       uint32_t index,
       const std::shared_ptr<dwio::common::BufferedInput>& input);
 
+  bool preDecompRowGroup(uint32_t index);
+
   // No-op in Parquet. All readers switch row groups at the same time, there is
   // no on-demand skipping to a new row group.
   void advanceFieldReader(
@@ -93,6 +95,7 @@ class StructColumnReader : public dwio::common::SelectiveStructColumnReader {
   // The level information for extracting nulls for 'this' from the
   // repdefs in a leaf PageReader.
   arrow::LevelInfo levelInfo_;
+  bool needPreDecomp = true;
 };
 
 } // namespace facebook::velox::parquet