diff --git a/velox/dwio/common/tests/utils/DataSetBuilder.cpp b/velox/dwio/common/tests/utils/DataSetBuilder.cpp
index 07b6b245003f..d42d8033677e 100644
--- a/velox/dwio/common/tests/utils/DataSetBuilder.cpp
+++ b/velox/dwio/common/tests/utils/DataSetBuilder.cpp
@@ -316,6 +316,36 @@ DataSetBuilder& DataSetBuilder::makeMapStringValues(
   return *this;
 }
 
+void DataSetBuilder::adjustTimestampToPrecision(
+    VectorPtr batch,
+    TimestampPrecision precision) {
+  auto type = batch->type();
+
+  if (type->kind() == TypeKind::TIMESTAMP) {
+    auto rawValues =
+        batch->asUnchecked<FlatVector<Timestamp>>()->mutableRawValues();
+    for (auto i = 0; i < batch->size(); ++i) {
+      if (batch->isNullAt(i)) {
+        continue;
+      }
+
+      rawValues[i].toPrecision(precision);
+    }
+  } else if (type->kind() == TypeKind::ROW) {
+    for (auto& child : batch->as<RowVector>()->children()) {
+      adjustTimestampToPrecision(child, precision);
+    }
+  }
+}
+
+DataSetBuilder& DataSetBuilder::adjustTimestampToPrecision(
+    TimestampPrecision precision) {
+  for (auto& batch : *batches_) {
+    adjustTimestampToPrecision(batch, precision);
+  }
+  return *this;
+}
+
 std::unique_ptr<std::vector<RowVectorPtr>> DataSetBuilder::build() {
   return std::move(batches_);
 }
diff --git a/velox/dwio/common/tests/utils/DataSetBuilder.h b/velox/dwio/common/tests/utils/DataSetBuilder.h
index 4893c28336f6..0ffb7e017b32 100644
--- a/velox/dwio/common/tests/utils/DataSetBuilder.h
+++ b/velox/dwio/common/tests/utils/DataSetBuilder.h
@@ -50,6 +50,11 @@ class DataSetBuilder {
   // groups. Tests skipping row groups based on row group stats.
   DataSetBuilder& withRowGroupSpecificData(int32_t numRowsPerGroup);
 
+  DataSetBuilder& adjustTimestampToPrecision(TimestampPrecision precision);
+  void adjustTimestampToPrecision(
+      VectorPtr batch,
+      TimestampPrecision precision);
+
   // Makes all data in 'batches_' after firstRow non-null. This finds a sampling
   // of non-null values from each column and replaces nulls in the column in
   // question with one of these. A column where only nulls are found in sampling
diff --git a/velox/dwio/parquet/reader/ParquetColumnReader.cpp b/velox/dwio/parquet/reader/ParquetColumnReader.cpp
index a87a295a787c..641c7147da25 100644
--- a/velox/dwio/parquet/reader/ParquetColumnReader.cpp
+++ b/velox/dwio/parquet/reader/ParquetColumnReader.cpp
@@ -75,10 +75,19 @@ std::unique_ptr<dwio::common::SelectiveColumnReader> ParquetColumnReader::build(
       return std::make_unique<BooleanColumnReader>(
           requestedType, fileType, params, scanSpec);
 
-    case TypeKind::TIMESTAMP:
-      return std::make_unique<TimestampColumnReader>(
-          requestedType, fileType, params, scanSpec);
-
+    case TypeKind::TIMESTAMP: {
+      auto& parquetFileType = static_cast<const ParquetTypeWithId&>(*fileType);
+      auto logicalTypeOpt = parquetFileType.logicalType_;
+      if (logicalTypeOpt.has_value() &&
+          logicalTypeOpt.value().__isset.TIMESTAMP &&
+          parquetFileType.parquetType_ == thrift::Type::INT64) {
+        return std::make_unique<TimestampInt64ColumnReader>(
+            requestedType, fileType, params, scanSpec);
+      } else {
+        return std::make_unique<TimestampInt96ColumnReader>(
+            requestedType, fileType, params, scanSpec);
+      }
+    }
     default:
       VELOX_FAIL(
           "buildReader unhandled type: " +
diff --git a/velox/dwio/parquet/reader/ParquetReader.cpp b/velox/dwio/parquet/reader/ParquetReader.cpp
index e36e305fe943..9d2accb7ea94 100644
--- a/velox/dwio/parquet/reader/ParquetReader.cpp
+++ b/velox/dwio/parquet/reader/ParquetReader.cpp
@@ -105,6 +105,9 @@ class ReaderBase {
   TypePtr convertType(
       const thrift::SchemaElement& schemaElement,
       const TypePtr& requestedType) const;
+      
+  thrift::LogicalType getTimestampLogicalType(
+      thrift::ConvertedType::type type) const;
 
   template <typename T>
   static std::shared_ptr<const RowType> createRowType(
@@ -551,13 +554,19 @@ std::unique_ptr<ParquetTypeWithId> ReaderBase::getParquetColumnInfo(
     int32_t precision =
         schemaElement.__isset.precision ? schemaElement.precision : 0;
     int32_t scale = schemaElement.__isset.scale ? schemaElement.scale : 0;
-    int32_t type_length =
+    int32_t typeLength =
         schemaElement.__isset.type_length ? schemaElement.type_length : 0;
     std::vector<std::unique_ptr<dwio::common::TypeWithId>> children;
-    const std::optional<thrift::LogicalType> logicalType_ =
+    std::optional<thrift::LogicalType> logicalType =
         schemaElement.__isset.logicalType
         ? std::optional<thrift::LogicalType>(schemaElement.logicalType)
         : std::nullopt;
+
+    if (veloxType->kind() == TypeKind::TIMESTAMP &&
+        schemaElement.type == thrift::Type::INT64 && !logicalType.has_value()) {
+      logicalType = getTimestampLogicalType(schemaElement.converted_type);
+    }
+
     auto leafTypePtr = std::make_unique<ParquetTypeWithId>(
         veloxType,
         std::move(children),
@@ -566,14 +575,14 @@ std::unique_ptr<ParquetTypeWithId> ReaderBase::getParquetColumnInfo(
         columnIdx++,
         name,
         schemaElement.type,
-        logicalType_,
+        logicalType,
         maxRepeat,
         maxDefine,
         isOptional,
         isRepeated,
         precision,
         scale,
-        type_length);
+        typeLength);
 
     if (schemaElement.repetition_type ==
         thrift::FieldRepetitionType::REPEATED) {
@@ -602,6 +611,29 @@ std::unique_ptr<ParquetTypeWithId> ReaderBase::getParquetColumnInfo(
   return nullptr;
 }
 
+thrift::LogicalType ReaderBase::getTimestampLogicalType(
+    thrift::ConvertedType::type convertedType) const {
+  thrift::TimestampType timestamp;
+  timestamp.__set_isAdjustedToUTC(true);
+  thrift::TimeUnit unit;
+
+  if (convertedType == thrift::ConvertedType::TIMESTAMP_MICROS) {
+    thrift::MicroSeconds micros;
+    unit.__set_MICROS(micros);
+  } else if (convertedType == thrift::ConvertedType::TIMESTAMP_MILLIS) {
+    thrift::MilliSeconds millis;
+    unit.__set_MILLIS(millis);
+  } else {
+    VELOX_NYI("{} Timestamp unit not supported.", convertedType);
+  }
+
+  timestamp.__set_unit(unit);
+  thrift::LogicalType logicalType;
+  logicalType.__set_TIMESTAMP(timestamp);
+
+  return logicalType;
+}
+
 TypePtr ReaderBase::convertType(
     const thrift::SchemaElement& schemaElement,
     const TypePtr& requestedType) const {
diff --git a/velox/dwio/parquet/reader/TimestampColumnReader.h b/velox/dwio/parquet/reader/TimestampColumnReader.h
index 11eb00e24286..80c7cf215b75 100644
--- a/velox/dwio/parquet/reader/TimestampColumnReader.h
+++ b/velox/dwio/parquet/reader/TimestampColumnReader.h
@@ -21,9 +21,9 @@
 
 namespace facebook::velox::parquet {
 
-class TimestampColumnReader : public IntegerColumnReader {
+class TimestampInt96ColumnReader : public IntegerColumnReader {
  public:
-  TimestampColumnReader(
+  TimestampInt96ColumnReader(
       const TypePtr& requestedType,
       std::shared_ptr<const dwio::common::TypeWithId> fileType,
       ParquetParams& params,
@@ -49,19 +49,7 @@ class TimestampColumnReader : public IntegerColumnReader {
       if (resultVector->isNullAt(i)) {
         continue;
       }
-      const auto timestamp = rawValues[i];
-      uint64_t nanos = timestamp.getNanos();
-      switch (timestampPrecision_) {
-        case TimestampPrecision::kMilliseconds:
-          nanos = nanos / 1'000'000 * 1'000'000;
-          break;
-        case TimestampPrecision::kMicroseconds:
-          nanos = nanos / 1'000 * 1'000;
-          break;
-        case TimestampPrecision::kNanoseconds:
-          break;
-      }
-      rawValues[i] = Timestamp(timestamp.getSeconds(), nanos);
+      rawValues[i].toPrecision(timestampPrecision_);
     }
   }
 
@@ -82,4 +70,186 @@ class TimestampColumnReader : public IntegerColumnReader {
   TimestampPrecision timestampPrecision_;
 };
 
+class TimestampInt64ColumnReader : public IntegerColumnReader {
+ public:
+  TimestampInt64ColumnReader(
+      const TypePtr& requestedType,
+      std::shared_ptr<const dwio::common::TypeWithId> fileType,
+      ParquetParams& params,
+      common::ScanSpec& scanSpec)
+      : IntegerColumnReader(requestedType, fileType, params, scanSpec),
+        timestampPrecision_(params.timestampPrecision()) {
+    auto& parquetFileType = static_cast<const ParquetTypeWithId&>(*fileType_);
+    auto logicalTypeOpt = parquetFileType.logicalType_;
+    VELOX_CHECK(logicalTypeOpt.has_value());
+
+    auto logicalType = logicalTypeOpt.value();
+    VELOX_CHECK(logicalType.__isset.TIMESTAMP);
+
+    if (!logicalType.TIMESTAMP.isAdjustedToUTC) {
+      VELOX_NYI("Only UTC adjusted Timestamp is supported.");
+    }
+
+    if (logicalType.TIMESTAMP.unit.__isset.MICROS) {
+      parquetTimestampPrecision_ = TimestampPrecision::kMicroseconds;
+    } else if (logicalType.TIMESTAMP.unit.__isset.MILLIS) {
+      parquetTimestampPrecision_ = TimestampPrecision::kMilliseconds;
+    } else {
+      VELOX_NYI("Nano Timestamp unit is not supported.");
+    }
+  }
+
+  bool hasBulkPath() const override {
+    return true;
+  }
+
+  void
+  processNulls(const bool isNull, const RowSet rows, const uint64_t* rawNulls) {
+    if (!rawNulls) {
+      return;
+    }
+    auto rawTs = values_->asMutable<Timestamp>();
+
+    returnReaderNulls_ = false;
+    anyNulls_ = !isNull;
+    allNull_ = isNull;
+    vector_size_t idx = 0;
+    for (vector_size_t i = 0; i < numValues_; i++) {
+      if (isNull) {
+        if (bits::isBitNull(rawNulls, i)) {
+          bits::setNull(rawResultNulls_, idx);
+          addOutputRow(rows[i]);
+          idx++;
+        }
+      } else {
+        if (!bits::isBitNull(rawNulls, i)) {
+          bits::setNull(rawResultNulls_, idx, false);
+          rawTs[idx] = rawTs[i];
+          addOutputRow(rows[i]);
+          idx++;
+        }
+      }
+    }
+  }
+
+  void processFilter(
+      const common::Filter* filter,
+      const RowSet rows,
+      const uint64_t* rawNulls) {
+    auto rawTs = values_->asMutable<Timestamp>();
+
+    returnReaderNulls_ = false;
+    anyNulls_ = false;
+    allNull_ = true;
+    vector_size_t idx = 0;
+    for (vector_size_t i = 0; i < numValues_; i++) {
+      if (rawNulls && bits::isBitNull(rawNulls, i)) {
+        if (filter->testNull()) {
+          bits::setNull(rawResultNulls_, idx);
+          addOutputRow(rows[i]);
+          anyNulls_ = true;
+          idx++;
+        }
+      } else {
+        if (filter->testTimestamp(rawTs[i])) {
+          if (rawNulls) {
+            bits::setNull(rawResultNulls_, idx, false);
+          }
+          rawTs[idx] = rawTs[i];
+          addOutputRow(rows[i]);
+          allNull_ = false;
+          idx++;
+        }
+      }
+    }
+  }
+
+  void getValues(RowSet rows, VectorPtr* result) override {
+    getFlatValues<Timestamp, Timestamp>(rows, result, requestedType_);
+  }
+
+  template <bool isDense>
+  void readHelper(common::Filter* filter, RowSet rows) {
+    dwio::common::ExtractToReader extractValues(this);
+    common::AlwaysTrue alwaysTrue;
+    dwio::common::ColumnVisitor<
+        int64_t,
+        common::AlwaysTrue,
+        decltype(extractValues),
+        isDense>
+        visitor(alwaysTrue, this, rows, extractValues);
+    readWithVisitor(rows, visitor);
+
+    auto tsValues =
+        AlignedBuffer::allocate<Timestamp>(numValues_, &memoryPool_);
+    auto rawTs = tsValues->asMutable<Timestamp>();
+    auto rawTsInt64 = values_->asMutable<int64_t>();
+
+    const auto rawNulls =
+        resultNulls() ? resultNulls()->as<uint64_t>() : nullptr;
+
+    Timestamp timestamp;
+    for (vector_size_t i = 0; i < numValues_; i++) {
+      if (!rawNulls || !bits::isBitNull(rawNulls, i)) {
+        if (parquetTimestampPrecision_ == TimestampPrecision::kMicroseconds) {
+          timestamp = Timestamp::fromMicros(rawTsInt64[i]);
+        } else {
+          timestamp = Timestamp::fromMillis(rawTsInt64[i]);
+        }
+        rawTs[i] = timestamp;
+        rawTs[i].toPrecision(timestampPrecision_);
+      }
+    }
+
+    values_ = tsValues;
+    rawValues_ = values_->asMutable<char>();
+    outputRows_.clear();
+
+    switch (
+        !filter ||
+                (filter->kind() == common::FilterKind::kIsNotNull && !rawNulls)
+            ? common::FilterKind::kAlwaysTrue
+            : filter->kind()) {
+      case common::FilterKind::kAlwaysTrue:
+        // Simply add all rows to output.
+        for (vector_size_t i = 0; i < numValues_; i++) {
+          addOutputRow(rows[i]);
+        }
+        break;
+      case common::FilterKind::kIsNull:
+        processNulls(true, rows, rawNulls);
+        break;
+      case common::FilterKind::kIsNotNull:
+        processNulls(false, rows, rawNulls);
+        break;
+      case common::FilterKind::kTimestampRange:
+      case common::FilterKind::kMultiRange:
+        processFilter(filter, rows, rawNulls);
+        break;
+      default:
+        VELOX_UNSUPPORTED("Unsupported filter.");
+    }
+  }
+
+  void read(
+      vector_size_t offset,
+      RowSet rows,
+      const uint64_t* /*incomingNulls*/) override {
+    auto& data = formatData_->as<ParquetData>();
+    prepareRead<int64_t>(offset, rows, nullptr);
+
+    bool isDense = rows.back() == rows.size() - 1;
+    if (isDense) {
+      readHelper<true>(scanSpec_->filter(), rows);
+    } else {
+      readHelper<false>(scanSpec_->filter(), rows);
+    }
+
+    readOffset_ += rows.back() + 1;
+  }
+
+ private:
+  TimestampPrecision parquetTimestampPrecision_;
+  TimestampPrecision timestampPrecision_;
+};
 } // namespace facebook::velox::parquet
diff --git a/velox/dwio/parquet/tests/examples/int64_micros_dictionary.parquet b/velox/dwio/parquet/tests/examples/int64_micros_dictionary.parquet
new file mode 100644
index 000000000000..5a8c72053979
Binary files /dev/null and b/velox/dwio/parquet/tests/examples/int64_micros_dictionary.parquet differ
diff --git a/velox/dwio/parquet/tests/examples/int64_micros_plain.parquet b/velox/dwio/parquet/tests/examples/int64_micros_plain.parquet
new file mode 100644
index 000000000000..1cfa7ae43595
Binary files /dev/null and b/velox/dwio/parquet/tests/examples/int64_micros_plain.parquet differ
diff --git a/velox/dwio/parquet/tests/examples/int64_millis_compatibility.parquet b/velox/dwio/parquet/tests/examples/int64_millis_compatibility.parquet
new file mode 100644
index 000000000000..d3c39e2c26ee
Binary files /dev/null and b/velox/dwio/parquet/tests/examples/int64_millis_compatibility.parquet differ
diff --git a/velox/dwio/parquet/tests/examples/int64_millis_dictionary.parquet b/velox/dwio/parquet/tests/examples/int64_millis_dictionary.parquet
new file mode 100644
index 000000000000..bd2b55553ca8
Binary files /dev/null and b/velox/dwio/parquet/tests/examples/int64_millis_dictionary.parquet differ
diff --git a/velox/dwio/parquet/tests/examples/int64_millis_plain.parquet b/velox/dwio/parquet/tests/examples/int64_millis_plain.parquet
new file mode 100644
index 000000000000..d214125e5e81
Binary files /dev/null and b/velox/dwio/parquet/tests/examples/int64_millis_plain.parquet differ
diff --git a/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp b/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp
index 0b27395f9a00..f50eaee29799 100644
--- a/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp
+++ b/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp
@@ -269,6 +269,82 @@ TEST_F(E2EFilterTest, timestampDictionary) {
       20);
 }
 
+TEST_F(E2EFilterTest, timestampINT64MillisDictionary) {
+  options_.enableDictionary = true;
+  options_.dataPageSize = 4 * 1024;
+  options_.parquetWriteTimestampUnit =
+      std::optional<TimestampUnit>(TimestampUnit::kMilli);
+  options_.parquetWriteTimestampTimeZone = "utc";
+
+  testWithTypes(
+      "timestamp_val_0:timestamp,"
+      "timestamp_val_1:timestamp",
+      [&]() {
+        dataSetBuilder_->adjustTimestampToPrecision(
+            TimestampPrecision::kMilliseconds);
+      },
+      true,
+      {"timestamp_val_0", "timestamp_val_1"},
+      1);
+}
+
+TEST_F(E2EFilterTest, timestampINT64MillisPlain) {
+  options_.enableDictionary = false;
+  options_.dataPageSize = 4 * 1024;
+  options_.parquetWriteTimestampUnit =
+      std::optional<TimestampUnit>(TimestampUnit::kMilli);
+  options_.parquetWriteTimestampTimeZone = "utc";
+
+  testWithTypes(
+      "timestamp_val_0:timestamp,"
+      "timestamp_val_1:timestamp",
+      [&]() {
+        dataSetBuilder_->adjustTimestampToPrecision(
+            TimestampPrecision::kMilliseconds);
+      },
+      true,
+      {"timestamp_val_0", "timestamp_val_1"},
+      1);
+}
+
+TEST_F(E2EFilterTest, timestampINT64MicrosDictionary) {
+  options_.enableDictionary = true;
+  options_.dataPageSize = 4 * 1024;
+  options_.parquetWriteTimestampUnit =
+      std::optional<TimestampUnit>(TimestampUnit::kMicro);
+  options_.parquetWriteTimestampTimeZone = "utc";
+
+  testWithTypes(
+      "timestamp_val_0:timestamp,"
+      "timestamp_val_1:timestamp",
+      [&]() {
+        dataSetBuilder_->adjustTimestampToPrecision(
+            TimestampPrecision::kMicroseconds);
+      },
+      true,
+      {"timestamp_val_0", "timestamp_val_1"},
+      1);
+}
+
+TEST_F(E2EFilterTest, timestampINT64MicrosPlain) {
+  options_.enableDictionary = false;
+  options_.dataPageSize = 4 * 1024;
+  options_.parquetWriteTimestampUnit =
+      std::optional<TimestampUnit>(TimestampUnit::kMicro);
+  options_.parquetWriteTimestampTimeZone = "utc";
+
+  testWithTypes(
+      "timestamp_val_0:timestamp,"
+      "timestamp_val_1:timestamp",
+      [&]() {
+        dataSetBuilder_->adjustTimestampToPrecision(
+            TimestampPrecision::kMicroseconds);
+      },
+      true,
+      {"timestamp_val_0", "timestamp_val_1"},
+      1);
+}
+
 TEST_F(E2EFilterTest, floatAndDoubleDirect) {
   options_.enableDictionary = false;
   options_.dataPageSize = 4 * 1024;
diff --git a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp
index 4261ee702249..62c0d57978ef 100644
--- a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp
+++ b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp
@@ -221,6 +221,88 @@ class ParquetTableScanTest : public HiveConnectorTestBase {
     writer->close();
   }
 
+  void testTimestampINT64(
+      const std::string& fileName,
+      TimestampPrecision precision) {
+    std::vector<Timestamp> rawDataMicros = {
+        Timestamp(0, 0),
+        Timestamp(0, 1000),
+        Timestamp(-1, 999999000),
+        Timestamp(0, 1000000),
+        Timestamp(-1, 999000000),
+        Timestamp(0, 1001000),
+        Timestamp(-1, 998999000),
+        Timestamp(0, 999000),
+        Timestamp(-1, 999001000),
+        Timestamp(1, 0),
+        Timestamp(-1, 0),
+        Timestamp(1, 1000),
+        Timestamp(-2, 999999000),
+        Timestamp(0, 99999000),
+        Timestamp(-1, 900001000)};
+
+    std::vector<Timestamp> rawDataMillis = {
+        Timestamp(0, 0),
+        Timestamp(0, 1000000),
+        Timestamp(-1, 999000000),
+        Timestamp(1, 0),
+        Timestamp(-1, 0),
+        Timestamp(1, 1000000),
+        Timestamp(-2, 999000000),
+        Timestamp(0, 999000000),
+        Timestamp(-1, 1000000),
+        Timestamp(1000, 0),
+        Timestamp(-1000, 0),
+        Timestamp(1000, 1000000),
+        Timestamp(-1001, 999000000),
+        Timestamp(99, 999000000),
+        Timestamp(-100, 1000000)};
+
+    auto a = makeFlatVector<Timestamp>(60, [&](auto row) {
+      return (precision == TimestampPrecision::kMilliseconds)
+          ? rawDataMillis[row / 4]
+          : rawDataMicros[row / 4];
+    });
+
+    auto expected = makeRowVector({"time"}, {a});
+    auto schema = asRowType(expected->type());
+    auto plan = PlanBuilder().tableScan(schema).planNode();
+
+    CursorParameters params;
+    std::shared_ptr<folly::Executor> executor =
+        std::make_shared<folly::CPUThreadPoolExecutor>(
+            std::thread::hardware_concurrency());
+    std::shared_ptr<core::QueryCtx> queryCtx =
+        core::QueryCtx::create(executor.get());
+    std::unordered_map<std::string, std::string> session = {
+        {std::string(connector::hive::HiveConfig::kReadTimestampUnitSession),
+         (precision == TimestampPrecision::kMilliseconds) ? "3" : "6"}};
+    queryCtx->setConnectorSessionOverridesUnsafe(
+        kHiveConnectorId, std::move(session));
+    params.queryCtx = queryCtx;
+    params.planNode = plan;
+    const int numSplitsPerFile = 1;
+
+    bool noMoreSplits = false;
+    auto addSplits = [&](exec::Task* task) {
+      if (!noMoreSplits) {
+        auto const splits = HiveConnectorTestBase::makeHiveConnectorSplits(
+            {getExampleFilePath(fileName)},
+            numSplitsPerFile,
+            dwio::common::FileFormat::PARQUET);
+        for (const auto& split : splits) {
+          task->addSplit("0", exec::Split(split));
+        }
+        task->noMoreSplits("0");
+      }
+      noMoreSplits = true;
+    };
+
+    auto result = readCursor(params, addSplits);
+    ASSERT_TRUE(waitForTaskCompletion(result.first->task().get()));
+    assertEqualResults({expected}, result.second);
+  }
+
  private:
   RowTypePtr getRowType(std::vector<std::string>&& outputColumnNames) const {
     std::vector<TypePtr> types;
@@ -836,6 +918,112 @@ TEST_F(ParquetTableScanTest, timestampPrecisionMicrosecond) {
   assertEqualResults({expected}, result.second);
 }
 
+TEST_F(ParquetTableScanTest, timestampINT64millis) {
+  testTimestampINT64(
+      "int64_millis_dictionary.parquet", TimestampPrecision::kMilliseconds);
+  testTimestampINT64(
+      "int64_millis_plain.parquet", TimestampPrecision::kMilliseconds);
+}
+
+TEST_F(ParquetTableScanTest, timestampINT64micros) {
+  testTimestampINT64(
+      "int64_micros_dictionary.parquet", TimestampPrecision::kMicroseconds);
+  testTimestampINT64(
+      "int64_micros_plain.parquet", TimestampPrecision::kMicroseconds);
+}
+
+TEST_F(ParquetTableScanTest, timestampINT64Filter) {
+  std::vector<Timestamp> rawDataMillis = {
+      Timestamp(0, 0),
+      Timestamp(0, 1000000),
+      Timestamp(-1, 999000000),
+      Timestamp(1, 0),
+      Timestamp(-1, 0),
+      Timestamp(1, 1000000),
+      Timestamp(-2, 999000000),
+      Timestamp(0, 999000000),
+      Timestamp(-1, 1000000),
+      Timestamp(1000, 0),
+      Timestamp(-1000, 0),
+      Timestamp(1000, 1000000),
+      Timestamp(-1001, 999000000),
+      Timestamp(99, 999000000),
+      Timestamp(-100, 1000000)};
+
+  auto a = makeFlatVector<Timestamp>(
+      60, [&](auto row) { return rawDataMillis[row / 4]; });
+
+  auto expected = makeRowVector({"time"}, {a});
+  createDuckDbTable("expected", {expected});
+
+  auto vector = makeArrayVector<Timestamp>({{}});
+
+  loadData(
+      getExampleFilePath("int64_millis_dictionary.parquet"),
+      ROW({"time"}, {TIMESTAMP()}),
+      makeRowVector(
+          {"time"},
+          {
+              vector,
+          }));
+
+  assertSelectWithFilter(
+      {"time"},
+      {},
+      "time < TIMESTAMP '1970-01-01 00:00:00'",
+      "SELECT time from expected where time < TIMESTAMP '1970-01-01 00:00:00'");
+
+  assertSelectWithFilter(
+      {"time"},
+      {},
+      "time <= TIMESTAMP '1970-01-01 00:00:00'",
+      "SELECT time from expected where time <= TIMESTAMP '1970-01-01 00:00:00'");
+
+  assertSelectWithFilter(
+      {"time"},
+      {},
+      "time > TIMESTAMP '1970-01-01 00:00:00'",
+      "SELECT time from expected where time > TIMESTAMP '1970-01-01 00:00:00'");
+
+  assertSelectWithFilter(
+      {"time"},
+      {},
+      "time >= TIMESTAMP '1970-01-01 00:00:00'",
+      "SELECT time from expected where time >= TIMESTAMP '1970-01-01 00:00:00'");
+
+  assertSelectWithFilter(
+      {"time"},
+      {},
+      "time == TIMESTAMP '1970-01-01 00:00:00'",
+      "SELECT time from expected where time == TIMESTAMP '1970-01-01 00:00:00'");
+
+  assertSelectWithFilter(
+      {"time"},
+      {},
+      "time != TIMESTAMP '1970-01-01 00:00:00'",
+      "SELECT time from expected where time != TIMESTAMP '1970-01-01 00:00:00'");
+}
+
+TEST_F(ParquetTableScanTest, timestampINT64BackwardCompatible) {
+  auto a = makeFlatVector<Timestamp>(
+      3, [](auto row) { return Timestamp(0, 10 * 1000000L); });
+
+  auto expected = makeRowVector({"time"}, {a});
+  createDuckDbTable("expected", {expected});
+
+  auto vector = makeArrayVector<Timestamp>({{}});
+
+  loadData(
+      getExampleFilePath("int64_millis_compatibility.parquet"),
+      ROW({"time"}, {TIMESTAMP()}),
+      makeRowVector(
+          {"time"},
+          {
+              vector,
+          }));
+  assertSelect({"time"}, "SELECT time from expected");
+}
+
 int main(int argc, char** argv) {
   testing::InitGoogleTest(&argc, argv);
   folly::Init init{&argc, &argv, false};
diff --git a/velox/type/Timestamp.h b/velox/type/Timestamp.h
index 44117bc7886b..57cec6ca5f3b 100644
--- a/velox/type/Timestamp.h
+++ b/velox/type/Timestamp.h
@@ -195,6 +195,19 @@ struct Timestamp {
     }
   }
 
+  void toPrecision(const TimestampPrecision& precision) {
+    switch (precision) {
+      case TimestampPrecision::kMilliseconds:
+        nanos_ = nanos_ / 1'000'000 * 1'000'000;
+        break;
+      case TimestampPrecision::kMicroseconds:
+        nanos_ = nanos_ / 1'000 * 1'000;
+        break;
+      case TimestampPrecision::kNanoseconds:
+        break;
+    }
+  }
+
   /// Exports the current timestamp as a std::chrono::time_point of millisecond
   /// precision. Note that the conversion may overflow since the internal
   /// `seconds_` value will need to be multiplied by 1000.