From 00a2d279bd841bf411c1a736f57713cb8e740eb1 Mon Sep 17 00:00:00 2001
From: jievince <38901892+jievince@users.noreply.github.com>
Date: Thu, 30 Sep 2021 00:00:11 +0800
Subject: [PATCH] Geo spatial: 3. geography data and index

---
 src/clients/meta/MetaClient.cpp             |  10 +-
 src/codec/RowReaderV2.cpp                   |  18 +-
 src/codec/RowWriterV2.cpp                   |  18 +-
 src/codec/RowWriterV2.h                     |   3 +
 src/codec/test/ResultSchemaProvider.cpp     |   8 +-
 src/codec/test/ResultSchemaProvider.h       |   5 +-
 src/codec/test/SchemaWriter.cpp             |   6 +-
 src/codec/test/SchemaWriter.h               |   3 +-
 src/common/datatypes/Geography.cpp          |  60 ++-
 src/common/datatypes/Geography.h            |  11 +-
 src/common/function/FunctionManager.cpp     |  39 +-
 src/common/geo/CMakeLists.txt               |   5 +
 src/common/geo/GeoFunction.cpp              |  72 ++-
 src/common/geo/GeoFunction.h                |  11 +-
 src/common/geo/GeoIndex.cpp                 | 190 +++++++
 src/common/geo/GeoIndex.h                   | 103 ++++
 src/common/geo/GeoUtils.h                   |  33 +-
 src/common/geo/test/GeoFunctionTest.cpp     |  64 +--
 src/common/meta/NebulaSchemaProvider.cpp    |   6 +-
 src/common/meta/NebulaSchemaProvider.h      |  12 +-
 src/common/meta/SchemaProviderIf.h          |   4 +
 src/common/utils/IndexKeyUtils.h            |  18 +-
 src/common/utils/test/CMakeLists.txt        |   3 +
 src/daemons/CMakeLists.txt                  |   1 +
 src/graph/util/SchemaUtil.cpp               |  10 +-
 src/graph/validator/LookupValidator.cpp     | 165 +++++-
 src/graph/validator/LookupValidator.h       |   4 +
 src/graph/validator/MaintainValidator.cpp   |   2 +
 src/kvstore/test/CMakeLists.txt             |   1 +
 src/meta/CMakeLists.txt                     |   1 +
 src/parser/test/ParserTest.cpp              |  21 +
 src/storage/index/LookupBaseProcessor-inl.h |   6 +-
 src/storage/test/CMakeLists.txt             |   1 +
 src/tools/db-dump/CMakeLists.txt            |   1 +
 src/tools/db-upgrade/CMakeLists.txt         |   1 +
 src/tools/meta-dump/CMakeLists.txt          |   1 +
 src/tools/simple-kv-verify/CMakeLists.txt   |   1 +
 src/tools/storage-perf/CMakeLists.txt       |   1 +
 tests/tck/features/geo/GeoBase.feature      | 526 ++++++++++++++++++++
 tests/tck/features/geo/GeoIndex.feature     |   0
 40 files changed, 1327 insertions(+), 118 deletions(-)
 create mode 100644 src/common/geo/GeoIndex.cpp
 create mode 100644 src/common/geo/GeoIndex.h
 create mode 100644 tests/tck/features/geo/GeoBase.feature
 create mode 100644 tests/tck/features/geo/GeoIndex.feature

diff --git a/src/clients/meta/MetaClient.cpp b/src/clients/meta/MetaClient.cpp
index d4766213d32..1d56abdce57 100644
--- a/src/clients/meta/MetaClient.cpp
+++ b/src/clients/meta/MetaClient.cpp
@@ -346,6 +346,8 @@ bool MetaClient::loadSchemas(GraphSpaceID spaceId,
     bool hasDef = col.default_value_ref().has_value();
     auto& colType = col.get_type();
     size_t len = colType.type_length_ref().has_value() ? *colType.get_type_length() : 0;
+    cpp2::GeoShape geoShape =
+        colType.geo_shape_ref().has_value() ? *colType.get_geo_shape() : cpp2::GeoShape::ANY;
     bool nullable = col.nullable_ref().has_value() ? *col.get_nullable() : false;
     Expression* defaultValueExpr = nullptr;
     if (hasDef) {
@@ -359,8 +361,12 @@ bool MetaClient::loadSchemas(GraphSpaceID spaceId,
       }
     }
 
-    schema->addField(
-        col.get_name(), colType.get_type(), len, nullable, hasDef ? defaultValueExpr : nullptr);
+    schema->addField(col.get_name(),
+                     colType.get_type(),
+                     len,
+                     nullable,
+                     hasDef ? defaultValueExpr : nullptr,
+                     geoShape);
   };
 
   for (auto& tagIt : tagItemVec) {
diff --git a/src/codec/RowReaderV2.cpp b/src/codec/RowReaderV2.cpp
index 0e8bb247c75..39442ac30ca 100644
--- a/src/codec/RowReaderV2.cpp
+++ b/src/codec/RowReaderV2.cpp
@@ -176,8 +176,22 @@ Value RowReaderV2::getValueByIndex(const int64_t index) const noexcept {
       return dt;
     }
     case meta::cpp2::PropertyType::GEOGRAPHY: {
-      // TODO(jie)
-      return Geography();
+      int32_t strOffset;
+      int32_t strLen;
+      memcpy(reinterpret_cast<void*>(&strOffset), &data_[offset], sizeof(int32_t));
+      memcpy(reinterpret_cast<void*>(&strLen), &data_[offset + sizeof(int32_t)], sizeof(int32_t));
+      if (static_cast<size_t>(strOffset) == data_.size() && strLen == 0) {
+        return Value::kEmpty;  // Is it ok to return Value::kEmpty?
+      }
+      CHECK_LT(strOffset, data_.size());
+      auto wkb = std::string(&data_[strOffset], strLen);
+      // Parse a geography from the wkb, normalize it and then verify its validity.
+      auto geogRet = Geography::fromWKB(wkb, true, true);
+      if (!geogRet.ok()) {
+        LOG(ERROR) << "Geography::fromWKB failed: " << geogRet.status();
+        return Value::kNullBadData;  // Is it ok to return Value::kNullBadData?
+      }
+      return std::move(geogRet).value();
     }
     case meta::cpp2::PropertyType::UNKNOWN:
       break;
diff --git a/src/codec/RowWriterV2.cpp b/src/codec/RowWriterV2.cpp
index 1953e63d29a..a7073e41393 100644
--- a/src/codec/RowWriterV2.cpp
+++ b/src/codec/RowWriterV2.cpp
@@ -127,7 +127,7 @@ RowWriterV2::RowWriterV2(RowReader& reader) : RowWriterV2(reader.getSchema()) {
         set(i, v.moveDateTime());
         break;
       case Value::Type::GEOGRAPHY:
-        // TODO(jie)
+        set(i, v.moveGeography());
         break;
       default:
         LOG(FATAL) << "Invalid data: " << v << ", type: " << v.typeName();
@@ -207,8 +207,7 @@ WriteResult RowWriterV2::setValue(ssize_t index, const Value& val) noexcept {
     case Value::Type::DATETIME:
       return write(index, val.getDateTime());
     case Value::Type::GEOGRAPHY:
-      // TODO(jie)
-      return WriteResult::TYPE_MISMATCH;
+      return write(index, val.getGeography());
     default:
       return WriteResult::TYPE_MISMATCH;
   }
@@ -762,6 +761,17 @@ WriteResult RowWriterV2::write(ssize_t index, const DateTime& v) noexcept {
   return WriteResult::SUCCEEDED;
 }
 
+WriteResult RowWriterV2::write(ssize_t index, const Geography& v) noexcept {
+  auto field = schema_->field(index);
+  auto geoShape = field->geoShape();
+  if (geoShape != meta::cpp2::GeoShape::ANY &&
+      folly::to<uint32_t>(geoShape) != folly::to<uint32_t>(v.shape())) {
+    return WriteResult::TYPE_MISMATCH;
+  }
+  std::string wkb = v.asWKB();
+  return write(index, folly::StringPiece(wkb));
+}
+
 WriteResult RowWriterV2::checkUnsetFields() noexcept {
   DefaultValueContext expCtx;
   for (size_t i = 0; i < schema_->getNumFields(); i++) {
@@ -802,7 +812,7 @@ WriteResult RowWriterV2::checkUnsetFields() noexcept {
             r = write(i, defVal.getDateTime());
             break;
           case Value::Type::GEOGRAPHY:
-            // TODO(jie)
+            r = write(i, defVal.getGeography());
             break;
           default:
             LOG(FATAL) << "Unsupported default value type: " << defVal.typeName()
diff --git a/src/codec/RowWriterV2.h b/src/codec/RowWriterV2.h
index d3bec72064f..16030a00afd 100644
--- a/src/codec/RowWriterV2.h
+++ b/src/codec/RowWriterV2.h
@@ -63,6 +63,7 @@ enum class WriteResult {
         TIMESTAMP       (8 bytes)
         DATE            (4 bytes)
         DATETIME        (15 bytes)
+        GEOGRAPHY       (8 bytes) *
 
   All except STRING typed properties are stored in-place. The STRING property
   stored the offset of the string content in the first 4 bytes and the length
@@ -188,6 +189,8 @@ class RowWriterV2 {
   WriteResult write(ssize_t index, const Date& v) noexcept;
   WriteResult write(ssize_t index, const Time& v) noexcept;
   WriteResult write(ssize_t index, const DateTime& v) noexcept;
+
+  WriteResult write(ssize_t index, const Geography& v) noexcept;
 };
 
 }  // namespace nebula
diff --git a/src/codec/test/ResultSchemaProvider.cpp b/src/codec/test/ResultSchemaProvider.cpp
index 359ca2bc80f..80b23ae1b26 100644
--- a/src/codec/test/ResultSchemaProvider.cpp
+++ b/src/codec/test/ResultSchemaProvider.cpp
@@ -23,14 +23,16 @@ ResultSchemaProvider::ResultSchemaField::ResultSchemaField(std::string name,
                                                            bool nullable,
                                                            int32_t offset,
                                                            size_t nullFlagPos,
-                                                           Expression* defaultValue)
+                                                           Expression* defaultValue,
+                                                           meta::cpp2::GeoShape geoShape)
     : name_(std::move(name)),
       type_(type),
       size_(size),
       nullable_(nullable),
       offset_(offset),
       nullFlagPos_(nullFlagPos),
-      defaultValue_(defaultValue) {}
+      defaultValue_(defaultValue),
+      geoShape_(geoShape) {}
 
 const char* ResultSchemaProvider::ResultSchemaField::name() const { return name_.c_str(); }
 
@@ -50,6 +52,8 @@ size_t ResultSchemaProvider::ResultSchemaField::offset() const { return offset_;
 
 size_t ResultSchemaProvider::ResultSchemaField::nullFlagPos() const { return nullFlagPos_; }
 
+meta::cpp2::GeoShape ResultSchemaProvider::ResultSchemaField::geoShape() const { return geoShape_; }
+
 /***********************************
  *
  * ResultSchemaProvider
diff --git a/src/codec/test/ResultSchemaProvider.h b/src/codec/test/ResultSchemaProvider.h
index f0a6bff44ca..4c51288163e 100644
--- a/src/codec/test/ResultSchemaProvider.h
+++ b/src/codec/test/ResultSchemaProvider.h
@@ -22,7 +22,8 @@ class ResultSchemaProvider : public meta::SchemaProviderIf {
                                bool nullable,
                                int32_t offset,
                                size_t nullFlagPos,
-                               Expression* defaultValue = nullptr);
+                               Expression* defaultValue = nullptr,
+                               meta::cpp2::GeoShape = meta::cpp2::GeoShape::ANY);
 
     const char* name() const override;
     meta::cpp2::PropertyType type() const override;
@@ -32,6 +33,7 @@ class ResultSchemaProvider : public meta::SchemaProviderIf {
     size_t size() const override;
     size_t offset() const override;
     size_t nullFlagPos() const override;
+    meta::cpp2::GeoShape geoShape() const override;
 
    private:
     std::string name_;
@@ -41,6 +43,7 @@ class ResultSchemaProvider : public meta::SchemaProviderIf {
     int32_t offset_;
     size_t nullFlagPos_;
     Expression* defaultValue_;
+    meta::cpp2::GeoShape geoShape_;
   };
 
  public:
diff --git a/src/codec/test/SchemaWriter.cpp b/src/codec/test/SchemaWriter.cpp
index d7c11ebe1eb..41e56bd22cd 100644
--- a/src/codec/test/SchemaWriter.cpp
+++ b/src/codec/test/SchemaWriter.cpp
@@ -17,7 +17,8 @@ SchemaWriter& SchemaWriter::appendCol(folly::StringPiece name,
                                       PropertyType type,
                                       int32_t fixedStrLen,
                                       bool nullable,
-                                      Expression* defaultValue) noexcept {
+                                      Expression* defaultValue,
+                                      meta::cpp2::GeoShape geoShape) noexcept {
   using folly::hash::SpookyHashV2;
   uint64_t hash = SpookyHashV2::Hash64(name.data(), name.size(), 0);
   DCHECK(nameIndex_.find(hash) == nameIndex_.end());
@@ -87,7 +88,8 @@ SchemaWriter& SchemaWriter::appendCol(folly::StringPiece name,
     nullFlagPos = numNullableFields_++;
   }
 
-  columns_.emplace_back(name.toString(), type, size, nullable, offset, nullFlagPos, defaultValue);
+  columns_.emplace_back(
+      name.toString(), type, size, nullable, offset, nullFlagPos, defaultValue, geoShape);
   nameIndex_.emplace(std::make_pair(hash, columns_.size() - 1));
 
   return *this;
diff --git a/src/codec/test/SchemaWriter.h b/src/codec/test/SchemaWriter.h
index 516d120f519..55c5dfb356d 100644
--- a/src/codec/test/SchemaWriter.h
+++ b/src/codec/test/SchemaWriter.h
@@ -21,7 +21,8 @@ class SchemaWriter : public ResultSchemaProvider {
                           meta::cpp2::PropertyType type,
                           int32_t fixedStrLen = 0,
                           bool nullable = false,
-                          Expression* defaultValue = nullptr) noexcept;
+                          Expression* defaultValue = nullptr,
+                          meta::cpp2::GeoShape geoShape = meta::cpp2::GeoShape::ANY) noexcept;
 
  private:
 };
diff --git a/src/common/datatypes/Geography.cpp b/src/common/datatypes/Geography.cpp
index 922e2525301..aef9a457599 100644
--- a/src/common/datatypes/Geography.cpp
+++ b/src/common/datatypes/Geography.cpp
@@ -19,20 +19,25 @@
 
 namespace nebula {
 
+constexpr double kMaxLongitude = 180.0;
+constexpr double kMaxLatitude = 90.0;
+
 void Coordinate::normalize() {
   // Reduce the x(longitude) to the range [-180, 180] degrees
   x = std::remainder(x, 360.0);
 
   // Reduce the y(latitude) to the range [-90, 90] degrees
   double tmp = remainder(y, 360.0);
-  if (tmp > 90.0) {
+  if (tmp > kMaxLatitude) {
     y = 180.0 - tmp;
-  } else if (tmp < -90.0) {
+  } else if (tmp < -kMaxLatitude) {
     y = -180.0 - tmp;
   }
 }
 
-bool Coordinate::isValid() const { return std::abs(x) <= 180.0 && std::abs(y) <= 90.0; }
+bool Coordinate::isValid() const {
+  return std::abs(x) <= kMaxLongitude && std::abs(y) <= kMaxLatitude;
+}
 
 void Point::normalize() {}
 
@@ -46,6 +51,7 @@ bool LineString::isValid() const {
     return false;
   }
   auto s2Region = geo::GeoUtils::s2RegionFromGeography(*this);
+  CHECK_NOTNULL(s2Region);
   return static_cast<S2Polyline*>(s2Region.get())->IsValid();
 }
 
@@ -67,6 +73,7 @@ bool Polygon::isValid() const {
     }
   }
   auto s2Region = geo::GeoUtils::s2RegionFromGeography(*this);
+  CHECK_NOTNULL(s2Region);
   return static_cast<S2Polygon*>(s2Region.get())->IsValid();
 }
 
@@ -91,6 +98,27 @@ StatusOr<Geography> Geography::fromWKT(const std::string& wkt,
   return geog;
 }
 
+StatusOr<Geography> Geography::fromWKB(const std::string& wkb,
+                                       bool needNormalize,
+                                       bool verifyValidity) {
+  auto geogRet = geo::WKBReader().read(wkb);
+  if (!geogRet.ok()) {
+    return geogRet;
+  }
+  auto geog = std::move(geogRet).value();
+  if (needNormalize) {
+    geog.normalize();
+  }
+  if (verifyValidity) {
+    if (!geog.isValid()) {
+      return Status::Error("Failed to parse an valid Geography instance from the wkb `%s'",
+                           wkb.c_str());
+    }
+  }
+
+  return geog;
+}
+
 GeoShape Geography::shape() const {
   switch (geo_.index()) {
     case 0:
@@ -184,6 +212,32 @@ bool Geography::isValid() const {
   }
 }
 
+Point Geography::centroid() const {
+  switch (shape()) {
+    case GeoShape::POINT: {
+      return this->point();
+    }
+    case GeoShape::LINESTRING: {
+      auto s2Region = geo::GeoUtils::s2RegionFromGeography(*this);
+      CHECK_NOTNULL(s2Region);
+      S2Point s2Point = static_cast<S2Polyline*>(s2Region.get())->GetCentroid();
+      return Point(geo::GeoUtils::coordinateFromS2Point(s2Point));
+    }
+    case GeoShape::POLYGON: {
+      auto s2Region = geo::GeoUtils::s2RegionFromGeography(*this);
+      CHECK_NOTNULL(s2Region);
+      S2Point s2Point = static_cast<S2Polygon*>(s2Region.get())->GetCentroid();
+      return Point(geo::GeoUtils::coordinateFromS2Point(s2Point));
+    }
+    case GeoShape::UNKNOWN:
+    default: {
+      LOG(ERROR)
+          << "Geography shapes other than Point/LineString/Polygon are not currently supported";
+      return Point();
+    }
+  }
+}
+
 std::string Geography::asWKT() const { return geo::WKTWriter().write(*this); }
 
 std::string Geography::asWKB() const { return geo::WKBWriter().write(*this); }
diff --git a/src/common/datatypes/Geography.h b/src/common/datatypes/Geography.h
index 5f2ecd31790..5e9b92de31c 100644
--- a/src/common/datatypes/Geography.h
+++ b/src/common/datatypes/Geography.h
@@ -58,8 +58,9 @@ struct Coordinate {
   }
   void __clear() { clear(); }
 
-  // TODO(jie) compare double correctly
-  bool operator==(const Coordinate& rhs) const { return x == rhs.x && y == rhs.y; }
+  bool operator==(const Coordinate& rhs) const {
+    return std::abs(x - rhs.x) < kEpsilon && std::abs(y - rhs.y) < kEpsilon;
+  }
   bool operator!=(const Coordinate& rhs) const { return !(*this == rhs); }
   bool operator<(const Coordinate& rhs) const {
     if (x != rhs.x) {
@@ -135,6 +136,10 @@ struct Geography {
                                      bool needNormalize = false,
                                      bool verifyValidity = false);
 
+  static StatusOr<Geography> fromWKB(const std::string& wkb,
+                                     bool needNormalize = false,
+                                     bool verifyValidity = false);
+
   Geography() {}
   Geography(const Point& v) : geo_(v) {}             // NOLINT
   Geography(Point&& v) : geo_(std::move(v)) {}       // NOLINT
@@ -156,6 +161,8 @@ struct Geography {
   void normalize();
   bool isValid() const;
 
+  Point centroid() const;
+
   std::string asWKT() const;
 
   std::string asWKB() const;
diff --git a/src/common/function/FunctionManager.cpp b/src/common/function/FunctionManager.cpp
index a25c38fdcf2..c35d2cb0f89 100644
--- a/src/common/function/FunctionManager.cpp
+++ b/src/common/function/FunctionManager.cpp
@@ -338,6 +338,11 @@ std::unordered_map<std::string, std::vector<TypeSignature>> FunctionManager::typ
      {
          TypeSignature({Value::Type::GEOGRAPHY}, Value::Type::STRING),
      }},
+    // geo transformations
+    {"st_centroid",
+     {
+         TypeSignature({Value::Type::GEOGRAPHY}, Value::Type::GEOGRAPHY),
+     }},
     // geo accessors
     {"st_isvalid",
      {
@@ -362,6 +367,11 @@ std::unordered_map<std::string, std::vector<TypeSignature>> FunctionManager::typ
      {
          TypeSignature({Value::Type::GEOGRAPHY, Value::Type::GEOGRAPHY, Value::Type::FLOAT},
                        Value::Type::BOOL),
+         TypeSignature({Value::Type::GEOGRAPHY,
+                        Value::Type::GEOGRAPHY,
+                        Value::Type::FLOAT,
+                        Value::Type::BOOL},
+                       Value::Type::BOOL),
      }},
     // geo measures
     {"st_distance",
@@ -2414,6 +2424,19 @@ FunctionManager::FunctionManager() {
       return g.asWKBHex();
     };
   }
+  // geo transformations
+  {
+    auto &attr = functions_["st_centroid"];
+    attr.minArity_ = 1;
+    attr.maxArity_ = 1;
+    attr.isPure_ = true;
+    attr.body_ = [](const auto &args) -> Value {
+      if (!args[0].get().isGeography()) {
+        return Value::kNullBadType;
+      }
+      return Geography(args[0].get().getGeography().centroid());
+    };
+  }
   // geo accessors
   {
     auto &attr = functions_["st_isvalid"];
@@ -2469,17 +2492,24 @@ FunctionManager::FunctionManager() {
   {
     auto &attr = functions_["st_dwithin"];
     attr.minArity_ = 3;
-    attr.maxArity_ = 3;
+    attr.maxArity_ = 4;
     attr.isPure_ = true;
     attr.body_ = [](const auto &args) -> Value {
       if (!args[0].get().isGeography() || !args[1].get().isGeography() ||
           !args[2].get().isFloat()) {
         return Value::kNullBadType;
       }
+      bool exclusive = false;
+      if (args.size() == 4) {
+        if (!args[3].get().isBool()) {
+          return Value::kNullBadType;
+        }
+        exclusive = args[3].get().getBool();
+      }
       return geo::GeoFunction::dWithin(args[0].get().getGeography(),
                                        args[1].get().getGeography(),
                                        args[2].get().getFloat(),
-                                       true);
+                                       exclusive);
     };
   }
   // geo measures
@@ -2515,6 +2545,11 @@ FunctionManager::FunctionManager() {
           return Value::kNullBadData;
         }
       }
+      const auto &geog = args[0].get().getGeography();
+      if (geog.shape() != GeoShape::POINT) {
+        LOG(ERROR) << "S2_CellIdFromPoint only accepts point argument";
+        return Value::kNullBadData;
+      }
       // TODO(jie) Should return uint64_t Value
       uint64_t cellId = geo::GeoFunction::s2CellIdFromPoint(args[0].get().getGeography(), level);
       const char *tmp = reinterpret_cast<const char *>(&cellId);
diff --git a/src/common/geo/CMakeLists.txt b/src/common/geo/CMakeLists.txt
index f7eafd34bb0..a9645de0833 100644
--- a/src/common/geo/CMakeLists.txt
+++ b/src/common/geo/CMakeLists.txt
@@ -3,5 +3,10 @@
 # This source code is licensed under Apache 2.0 License,
 # attached with Common Clause Condition 1.0, found in the LICENSES directory.
 
+nebula_add_library(
+    geo_index_obj OBJECT
+    GeoIndex.cpp
+)
+
 nebula_add_subdirectory(io)
 nebula_add_subdirectory(test)
diff --git a/src/common/geo/GeoFunction.cpp b/src/common/geo/GeoFunction.cpp
index 84bdc7ecb6d..5ceedb47b54 100644
--- a/src/common/geo/GeoFunction.cpp
+++ b/src/common/geo/GeoFunction.cpp
@@ -23,7 +23,7 @@ namespace geo {
 bool GeoFunction::intersects(const Geography& a, const Geography& b) {
   auto aRegion = a.asS2();
   auto bRegion = b.asS2();
-  if (!aRegion || !bRegion) {
+  if (UNLIKELY(!aRegion || !bRegion)) {
     return false;
   }
 
@@ -99,7 +99,7 @@ bool GeoFunction::intersects(const Geography& a, const Geography& b) {
 bool GeoFunction::covers(const Geography& a, const Geography& b) {
   auto aRegion = a.asS2();
   auto bRegion = b.asS2();
-  if (!aRegion || !bRegion) {
+  if (UNLIKELY(!aRegion || !bRegion)) {
     return false;
   }
 
@@ -178,10 +178,10 @@ bool GeoFunction::covers(const Geography& a, const Geography& b) {
 
 bool GeoFunction::coveredBy(const Geography& a, const Geography& b) { return covers(b, a); }
 
-bool GeoFunction::dWithin(const Geography& a, const Geography& b, double distance, bool inclusive) {
+bool GeoFunction::dWithin(const Geography& a, const Geography& b, double distance, bool exclusive) {
   auto aRegion = a.asS2();
   auto bRegion = b.asS2();
-  if (!aRegion || !bRegion) {
+  if (UNLIKELY(!aRegion || !bRegion)) {
     return false;
   }
 
@@ -192,15 +192,15 @@ bool GeoFunction::dWithin(const Geography& a, const Geography& b, double distanc
         case GeoShape::POINT: {
           const S2Point& bPoint = static_cast<S2PointRegion*>(bRegion.get())->point();
           double closestDistance = S2Earth::GetDistanceMeters(aPoint, bPoint);
-          return inclusive ? closestDistance <= distance : closestDistance < distance;
+          return exclusive ? closestDistance < distance : closestDistance <= distance;
         }
         case GeoShape::LINESTRING: {
           S2Polyline* bLine = static_cast<S2Polyline*>(bRegion.get());
-          return s2PointAndS2PolylineAreWithinDistance(aPoint, bLine, distance, inclusive);
+          return s2PointAndS2PolylineAreWithinDistance(aPoint, bLine, distance, exclusive);
         }
         case GeoShape::POLYGON: {
           S2Polygon* bPolygon = static_cast<S2Polygon*>(bRegion.get());
-          return s2PointAndS2PolygonAreWithinDistance(aPoint, bPolygon, distance, inclusive);
+          return s2PointAndS2PolygonAreWithinDistance(aPoint, bPolygon, distance, exclusive);
         }
         case GeoShape::UNKNOWN:
         default: {
@@ -215,7 +215,7 @@ bool GeoFunction::dWithin(const Geography& a, const Geography& b, double distanc
       switch (b.shape()) {
         case GeoShape::POINT: {
           const S2Point& bPoint = static_cast<S2PointRegion*>(bRegion.get())->point();
-          return s2PointAndS2PolylineAreWithinDistance(bPoint, aLine, distance, inclusive);
+          return s2PointAndS2PolylineAreWithinDistance(bPoint, aLine, distance, exclusive);
         }
         case GeoShape::LINESTRING: {
           S2Polyline* bLine = static_cast<S2Polyline*>(bRegion.get());
@@ -224,16 +224,16 @@ bool GeoFunction::dWithin(const Geography& a, const Geography& b, double distanc
           bIndex.Add(std::make_unique<S2Polyline::Shape>(bLine));
           S2ClosestEdgeQuery query(&aIndex);
           S2ClosestEdgeQuery::ShapeIndexTarget target(&bIndex);
-          if (inclusive) {
-            return query.IsDistanceLessOrEqual(
-                &target, S2Earth::ToChordAngle(util::units::Meters(distance)));
+          if (exclusive) {
+            return query.IsDistanceLess(&target,
+                                        S2Earth::ToChordAngle(util::units::Meters(distance)));
           }
-          return query.IsDistanceLess(&target,
-                                      S2Earth::ToChordAngle(util::units::Meters(distance)));
+          return query.IsDistanceLessOrEqual(&target,
+                                             S2Earth::ToChordAngle(util::units::Meters(distance)));
         }
         case GeoShape::POLYGON: {
           S2Polygon* bPolygon = static_cast<S2Polygon*>(bRegion.get());
-          return s2PolylineAndS2PolygonAreWithinDistance(aLine, bPolygon, distance, inclusive);
+          return s2PolylineAndS2PolygonAreWithinDistance(aLine, bPolygon, distance, exclusive);
         }
         case GeoShape::UNKNOWN:
         default: {
@@ -248,22 +248,22 @@ bool GeoFunction::dWithin(const Geography& a, const Geography& b, double distanc
       switch (b.shape()) {
         case GeoShape::POINT: {
           const S2Point& bPoint = static_cast<S2PointRegion*>(bRegion.get())->point();
-          return s2PointAndS2PolygonAreWithinDistance(bPoint, aPolygon, distance, inclusive);
+          return s2PointAndS2PolygonAreWithinDistance(bPoint, aPolygon, distance, exclusive);
         }
         case GeoShape::LINESTRING: {
           S2Polyline* bLine = static_cast<S2Polyline*>(bRegion.get());
-          return s2PolylineAndS2PolygonAreWithinDistance(bLine, aPolygon, distance, inclusive);
+          return s2PolylineAndS2PolygonAreWithinDistance(bLine, aPolygon, distance, exclusive);
         }
         case GeoShape::POLYGON: {
           S2Polygon* bPolygon = static_cast<S2Polygon*>(bRegion.get());
           S2ClosestEdgeQuery query(&aPolygon->index());
           S2ClosestEdgeQuery::ShapeIndexTarget target(&bPolygon->index());
-          if (inclusive) {
-            return query.IsDistanceLessOrEqual(
-                &target, S2Earth::ToChordAngle(util::units::Meters(distance)));
+          if (exclusive) {
+            return query.IsDistanceLess(&target,
+                                        S2Earth::ToChordAngle(util::units::Meters(distance)));
           }
-          return query.IsDistanceLess(&target,
-                                      S2Earth::ToChordAngle(util::units::Meters(distance)));
+          return query.IsDistanceLessOrEqual(&target,
+                                             S2Earth::ToChordAngle(util::units::Meters(distance)));
         }
         case GeoShape::UNKNOWN:
         default: {
@@ -287,7 +287,7 @@ bool GeoFunction::dWithin(const Geography& a, const Geography& b, double distanc
 double GeoFunction::distance(const Geography& a, const Geography& b) {
   auto aRegion = a.asS2();
   auto bRegion = b.asS2();
-  if (!aRegion || !bRegion) {
+  if (UNLIKELY(!aRegion || !bRegion)) {
     return -1.0;
   }
 
@@ -381,7 +381,7 @@ double GeoFunction::distance(const Geography& a, const Geography& b) {
 
 uint64_t GeoFunction::s2CellIdFromPoint(const Geography& a, int level) {
   auto aRegion = a.asS2();
-  if (!aRegion) {
+  if (UNLIKELY(!aRegion)) {
     return -1;
   }
   if (level < 0 || level > 30) {
@@ -410,7 +410,7 @@ uint64_t GeoFunction::s2CellIdFromPoint(const Geography& a, int level) {
 std::vector<uint64_t> GeoFunction::s2CoveringCellIds(
     const Geography& a, int minLevel, int maxLevel, int maxCells, double bufferInMeters) {
   auto aRegion = a.asS2();
-  if (!aRegion) {
+  if (UNLIKELY(!aRegion)) {
     return {};
   }
   if (minLevel < 0 || minLevel > 30) {
@@ -507,47 +507,41 @@ double GeoFunction::distanceOfS2PolygonWithS2Point(const S2Polygon* aPolygon,
 bool GeoFunction::s2PointAndS2PolylineAreWithinDistance(const S2Point& aPoint,
                                                         const S2Polyline* bLine,
                                                         double distance,
-                                                        bool inclusive) {
+                                                        bool exclusive) {
   MutableS2ShapeIndex bIndex;
   bIndex.Add(std::make_unique<S2Polyline::Shape>(bLine));
   S2ClosestEdgeQuery query(&bIndex);
   S2ClosestEdgeQuery::PointTarget target(aPoint);
-  if (inclusive) {
-    return query.IsDistanceLessOrEqual(&target,
-                                       S2Earth::ToChordAngle(util::units::Meters(distance)));
-  } else {
+  if (exclusive) {
     return query.IsDistanceLess(&target, S2Earth::ToChordAngle(util::units::Meters(distance)));
   }
+  return query.IsDistanceLessOrEqual(&target, S2Earth::ToChordAngle(util::units::Meters(distance)));
 }
 
 bool GeoFunction::s2PointAndS2PolygonAreWithinDistance(const S2Point& aPoint,
                                                        const S2Polygon* bPolygon,
                                                        double distance,
-                                                       bool inclusive) {
+                                                       bool exclusive) {
   S2ClosestEdgeQuery query(&bPolygon->index());
   S2ClosestEdgeQuery::PointTarget target(aPoint);
-  if (inclusive) {
-    return query.IsDistanceLessOrEqual(&target,
-                                       S2Earth::ToChordAngle(util::units::Meters(distance)));
-  } else {
+  if (exclusive) {
     return query.IsDistanceLess(&target, S2Earth::ToChordAngle(util::units::Meters(distance)));
   }
+  return query.IsDistanceLessOrEqual(&target, S2Earth::ToChordAngle(util::units::Meters(distance)));
 }
 
 bool GeoFunction::s2PolylineAndS2PolygonAreWithinDistance(const S2Polyline* aLine,
                                                           const S2Polygon* bPolygon,
                                                           double distance,
-                                                          bool inclusive) {
+                                                          bool exclusive) {
   MutableS2ShapeIndex aIndex;
   aIndex.Add(std::make_unique<S2Polyline::Shape>(aLine));
   S2ClosestEdgeQuery::ShapeIndexTarget target(&aIndex);
   S2ClosestEdgeQuery query(&bPolygon->index());
-  if (inclusive) {
-    return query.IsDistanceLessOrEqual(&target,
-                                       S2Earth::ToChordAngle(util::units::Meters(distance)));
-  } else {
+  if (exclusive) {
     return query.IsDistanceLess(&target, S2Earth::ToChordAngle(util::units::Meters(distance)));
   }
+  return query.IsDistanceLessOrEqual(&target, S2Earth::ToChordAngle(util::units::Meters(distance)));
 }
 
 }  // namespace geo
diff --git a/src/common/geo/GeoFunction.h b/src/common/geo/GeoFunction.h
index 5b37a998c15..7480a3187e4 100644
--- a/src/common/geo/GeoFunction.h
+++ b/src/common/geo/GeoFunction.h
@@ -29,7 +29,10 @@ class GeoFunction {
   // Returns true if any of a is within distance meters of b.
   // We don't need to find the closest points. We just need to find the first point pair whose
   // distance is less than or less equal than the given distance. (Early quit)
-  static bool dWithin(const Geography& a, const Geography& b, double distance, bool inclusive);
+  static bool dWithin(const Geography& a,
+                      const Geography& b,
+                      double distance,
+                      bool exclusive = false);
 
   // Return the closest distance in meters of a and b.
   static double distance(const Geography& a, const Geography& b);
@@ -56,17 +59,17 @@ class GeoFunction {
   static bool s2PointAndS2PolylineAreWithinDistance(const S2Point& aPoint,
                                                     const S2Polyline* bLine,
                                                     double distance,
-                                                    bool inclusive);
+                                                    bool exclusive);
 
   static bool s2PointAndS2PolygonAreWithinDistance(const S2Point& aPoint,
                                                    const S2Polygon* bPolygon,
                                                    double distance,
-                                                   bool inclusive);
+                                                   bool exclusive);
 
   static bool s2PolylineAndS2PolygonAreWithinDistance(const S2Polyline* aLine,
                                                       const S2Polygon* bPolygon,
                                                       double distance,
-                                                      bool inclusive);
+                                                      bool exclusive);
 };
 
 }  // namespace geo
diff --git a/src/common/geo/GeoIndex.cpp b/src/common/geo/GeoIndex.cpp
new file mode 100644
index 00000000000..24a279d57e4
--- /dev/null
+++ b/src/common/geo/GeoIndex.cpp
@@ -0,0 +1,190 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This sourc_e code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "common/geo/GeoIndex.h"
+
+#include <folly/String.h>
+#include <folly/hash/Hash.h>
+#include <s2/mutable_s2shape_index.h>
+#include <s2/s2cap.h>
+#include <s2/s2cell.h>
+#include <s2/s2cell_id.h>
+#include <s2/s2earth.h>
+#include <s2/s2latlng.h>
+#include <s2/s2polygon.h>
+#include <s2/s2region_coverer.h>
+#include <s2/s2shape_index_buffered_region.h>
+#include <s2/util/units/length-units.h>
+
+#include <cstdint>
+
+#include "common/datatypes/Geography.h"
+#include "common/utils/IndexKeyUtils.h"
+#include "interface/gen-cpp2/storage_types.h"
+
+namespace nebula {
+namespace geo {
+
+nebula::storage::cpp2::IndexColumnHint ScanRange::toIndexColumnHint() {
+  nebula::storage::cpp2::IndexColumnHint hint;
+  // column_name should be set by the caller
+  if (isRangeScan) {
+    hint.set_scan_type(nebula::storage::cpp2::ScanType::RANGE);
+    hint.set_begin_value(
+        IndexKeyUtils::encodeUint64(rangeMin));  // Encode uint64_t as string ahead of time
+    hint.set_end_value(IndexKeyUtils::encodeUint64(rangeMax));
+  } else {
+    hint.set_scan_type(nebula::storage::cpp2::ScanType::PREFIX);
+    hint.set_begin_value(IndexKeyUtils::encodeUint64(rangeMin));
+  }
+  return hint;
+}
+
+std::vector<uint64_t> GeoIndex::indexCells(const Geography& g) const noexcept {
+  auto cells = coveringCells(g);
+  std::vector<uint64_t> cellIds;
+  cellIds.reserve(cells.size());
+  for (auto& cell : cells) {
+    cellIds.push_back(cell.id());
+  }
+  return cellIds;
+}
+
+std::vector<ScanRange> GeoIndex::intersects(const Geography& g) const noexcept {
+  auto r = g.asS2();
+  if (UNLIKELY(!r)) {
+    return {};
+  }
+
+  return intersects(*r);
+}
+
+// covers degenerates to intersects currently
+std::vector<ScanRange> GeoIndex::covers(const Geography& g) const noexcept { return intersects(g); }
+
+// coveredBy degenerates to intersects currently
+std::vector<ScanRange> GeoIndex::coveredBy(const Geography& g) const noexcept {
+  return intersects(g);
+}
+
+std::vector<ScanRange> GeoIndex::dWithin(const Geography& g, double distance) const noexcept {
+  auto r = g.asS2();
+  if (UNLIKELY(!r)) {
+    return {};
+  }
+
+  S1Angle radius = S2Earth::ToAngle(util::units::Meters(distance));
+  // First expand the region, then build the covering
+  switch (g.shape()) {
+    case GeoShape::POINT: {
+      const S2Point& gPoint = static_cast<S2PointRegion*>(r.get())->point();
+      S2Cap gCap(gPoint, radius);
+      return intersects(gCap);
+    }
+    case GeoShape::LINESTRING: {
+      S2Polyline* gLine = static_cast<S2Polyline*>(r.get());
+      MutableS2ShapeIndex index;
+      index.Add(std::make_unique<S2Polyline::Shape>(gLine));
+      S2ShapeIndexBufferedRegion gBuffer(&index, radius);
+      return intersects(gBuffer);
+    }
+    case GeoShape::POLYGON: {
+      S2Polygon* gPolygon = static_cast<S2Polygon*>(r.get());
+      S2ShapeIndexBufferedRegion gBuffer(&gPolygon->index(), radius);
+      return intersects(gBuffer);
+    }
+    default:
+      LOG(FATAL)
+          << "Geography shapes other than Point/LineString/Polygon are not currently supported";
+      return {};
+  }
+}
+
+std::vector<ScanRange> GeoIndex::intersects(const S2Region& r) const noexcept {
+  auto cells = coveringCells(r);
+  std::vector<ScanRange> scanRanges;
+  for (const S2CellId& cellId : cells) {
+    if (cellId.is_leaf()) {
+      scanRanges.emplace_back(cellId.id());
+    } else {
+      scanRanges.emplace_back(cellId.range_min().id(), cellId.range_max().id());
+    }
+  }
+
+  // For the indexed column which only contains point, we don't need to get the ancestor cells,
+  // because the point is at max level(30).
+  if (!pointsOnly_) {
+    auto ancestors = ancestorCells(cells);
+    for (const S2CellId& cellId : ancestors) {
+      scanRanges.emplace_back(cellId.id());
+    }
+  }
+
+  return scanRanges;
+}
+
+std::vector<S2CellId> GeoIndex::coveringCells(const Geography& g) const noexcept {
+  auto r = g.asS2();
+  if (UNLIKELY(!r)) {
+    return {};
+  }
+
+  // Currently we don't apply region coverer params to point, because it's useless.
+  // Point always use level 30.
+  if (g.shape() == GeoShape::POINT) {
+    const S2Point& gPoint = static_cast<S2PointRegion*>(r.get())->point();
+    return {S2CellId(gPoint)};
+  }
+
+  S2RegionCoverer rc(rcParams_.s2RegionCovererOpts());
+  std::vector<S2CellId> covering;
+  rc.GetCovering(*r, &covering);
+  // 1. NO NEED TO CALL S2RegionCoverer::CanonicalizeCovering(covering), because the covering is
+  // already canonical, which means that is sorted, non-overlapping and satisfy the desired
+  // constraints min_level, max_level.
+  // 2. DO NOT CALL S2CellUnion::Normalize(covering), it will replacing groups of 4 child cells by
+  // their parent cell, In this case, it may cause the covering don't satisfy the desired
+  // constraints min_level.
+  return covering;
+}
+
+std::vector<S2CellId> GeoIndex::coveringCells(const S2Region& r) const noexcept {
+  S2RegionCoverer rc(rcParams_.s2RegionCovererOpts());
+  std::vector<S2CellId> covering;
+  rc.GetCovering(r, &covering);
+  return covering;
+}
+
+std::vector<S2CellId> GeoIndex::ancestorCells(const std::vector<S2CellId>& cells) const noexcept {
+  // DCHECK(rc.IsCanonical(cells));
+  std::vector<S2CellId> ancestors;
+  std::unordered_set<S2CellId> seen;
+  for (const auto& cellId : cells) {
+    for (auto l = cellId.level() - 1; l >= rcParams_.minCellLevel_; --l) {
+      S2CellId parentCellId = cellId.parent(l);
+      if (seen.find(parentCellId) != seen.end()) {
+        break;
+      }
+      seen.emplace(parentCellId);
+      ancestors.push_back(parentCellId);
+    }
+  }
+  // The ancestors here is non-overlapping but unsorted. Do we need to sort it?
+  // May need to call S2RegionCoverer::CanonicalizeCovering(&ancestors)?
+  return ancestors;
+}
+
+}  // namespace geo
+}  // namespace nebula
+
+namespace std {
+
+// Inject a customized hash function
+std::size_t hash<S2CellId>::operator()(const S2CellId& c) const noexcept {
+  return hash<uint64_t>{}(c.id());
+}
+
+}  // namespace std
diff --git a/src/common/geo/GeoIndex.h b/src/common/geo/GeoIndex.h
new file mode 100644
index 00000000000..fa79483a8d9
--- /dev/null
+++ b/src/common/geo/GeoIndex.h
@@ -0,0 +1,103 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#pragma once
+
+#include <s2/s2region_coverer.h>
+
+#include <algorithm>
+#include <vector>
+
+#include "common/datatypes/Geography.h"
+
+namespace nebula {
+
+namespace storage {
+namespace cpp2 {
+class IndexColumnHint;
+}  // namespace cpp2
+}  // namespace storage
+
+namespace geo {
+
+struct RegionCoverParams {
+  // TODO(jie): Find the best default params
+  int minCellLevel_ = 4;
+  int maxCellLevel_ = 23;  // About 1m
+  int maxCellNum_ = 8;
+
+  RegionCoverParams() = default;
+
+  RegionCoverParams(int minLevl, int maxLevl, int maxCells)
+      : minCellLevel_(minLevl), maxCellLevel_(maxLevl), maxCellNum_(maxCells) {}
+
+  S2RegionCoverer::Options s2RegionCovererOpts() const {
+    S2RegionCoverer::Options opts;
+    opts.set_min_level(minCellLevel_);
+    opts.set_max_level(maxCellLevel_);
+    opts.set_max_cells(maxCellNum_);
+    return opts;
+  }
+};
+
+// scan type: PREFIX or RANGE
+struct ScanRange {
+  uint64_t rangeMin;
+  uint64_t rangeMax;
+  bool isRangeScan;
+
+  ScanRange(uint64_t min, uint64_t max) : rangeMin(min), rangeMax(max), isRangeScan(true) {}
+
+  explicit ScanRange(uint64_t v) : rangeMin(v), isRangeScan(false) {}
+
+  nebula::storage::cpp2::IndexColumnHint toIndexColumnHint();
+};
+
+class GeoIndex {
+ public:
+  explicit GeoIndex(const RegionCoverParams& params, bool pointsOnly = false)
+      : rcParams_(params), pointsOnly_(pointsOnly) {}
+
+  // Build geo index for the geography g
+  std::vector<uint64_t> indexCells(const Geography& g) const noexcept;
+
+  // Query the geo index
+  // ST_Intersects(g, x), x is the indexed geography column
+  std::vector<ScanRange> intersects(const Geography& g) const noexcept;
+  // ST_Covers(g, x), x is the indexed geography column
+  std::vector<ScanRange> covers(const Geography& g) const noexcept;
+  // ST_CoveredBy(g, x), x is the indexed geography column
+  std::vector<ScanRange> coveredBy(const Geography& g) const noexcept;
+  // ST_Distance(g, x, distance), x is the indexed geography column
+  std::vector<ScanRange> dWithin(const Geography& g, double distance) const noexcept;
+
+ private:
+  std::vector<ScanRange> intersects(const S2Region& r) const noexcept;
+
+  std::vector<S2CellId> coveringCells(const Geography& g) const noexcept;
+
+  std::vector<S2CellId> coveringCells(const S2Region& r) const noexcept;
+
+  std::vector<S2CellId> ancestorCells(const std::vector<S2CellId>& cells) const noexcept;
+
+ private:
+  RegionCoverParams rcParams_;
+  bool pointsOnly_{
+      false};  // For the column Geography(Point), we don't need to build ancestor cells
+};
+
+}  // namespace geo
+}  // namespace nebula
+
+namespace std {
+
+// Inject a customized hash function
+template <>
+struct hash<S2CellId> {
+  std::size_t operator()(const S2CellId& h) const noexcept;
+};
+
+}  // namespace std
diff --git a/src/common/geo/GeoUtils.h b/src/common/geo/GeoUtils.h
index 7424399f76c..2136c687116 100644
--- a/src/common/geo/GeoUtils.h
+++ b/src/common/geo/GeoUtils.h
@@ -6,6 +6,7 @@
 
 #pragma once
 
+#include <s2/s2latlng.h>
 #include <s2/s2loop.h>
 #include <s2/s2polygon.h>
 
@@ -26,28 +27,23 @@ class GeoUtils final {
       }
       case GeoShape::LINESTRING: {
         const auto& lineString = geog.lineString();
-        auto coordList = lineString.coordList;
+        const auto& coordList = lineString.coordList;
         auto s2Points = s2PointsFromCoordinateList(coordList);
-        auto s2Polyline = std::make_unique<S2Polyline>(s2Points, S2Debug::DISABLE);
-        return s2Polyline;
+        return std::make_unique<S2Polyline>(s2Points, S2Debug::DISABLE);
       }
       case GeoShape::POLYGON: {
         const auto& polygon = geog.polygon();
         uint32_t numCoordList = polygon.numCoordList();
         std::vector<std::unique_ptr<S2Loop>> s2Loops;
         s2Loops.reserve(numCoordList);
-        for (size_t i = 0; i < numCoordList; ++i) {
-          auto coordList = polygon.coordListList[i];
-          if (!coordList.empty()) {
-            coordList.pop_back();  // Remove redundant last coordinate
-          }
-          auto s2Points = s2PointsFromCoordinateList(coordList);
+        for (const auto& coordList : polygon.coordListList) {
+          auto s2Points = s2PointsFromCoordinateList(
+              coordList, true);  // S2 doesn't need the redundant last point
           auto s2Loop = std::make_unique<S2Loop>(std::move(s2Points), S2Debug::DISABLE);
           s2Loop->Normalize();  // All loops must be oriented CCW(counterclockwise) for S2
           s2Loops.emplace_back(std::move(s2Loop));
         }
-        auto s2Polygon = std::make_unique<S2Polygon>(std::move(s2Loops), S2Debug::DISABLE);
-        return s2Polygon;
+        return std::make_unique<S2Polygon>(std::move(s2Loops), S2Debug::DISABLE);
       }
       default:
         LOG(FATAL)
@@ -62,9 +58,22 @@ class GeoUtils final {
     return latlng.ToPoint();
   }
 
-  static std::vector<S2Point> s2PointsFromCoordinateList(const std::vector<Coordinate>& coordList) {
+  static Coordinate coordinateFromS2Point(const S2Point& s2Point) {
+    S2LatLng s2Latlng(s2Point);
+    return Coordinate(s2Latlng.lat().degrees(), s2Latlng.lng().degrees());
+  }
+
+  static std::vector<S2Point> s2PointsFromCoordinateList(const std::vector<Coordinate>& coordList,
+                                                         bool excludeTheLast = false) {
     std::vector<S2Point> s2Points;
     uint32_t numCoords = coordList.size();
+    if (excludeTheLast) {
+      numCoords -= 1;
+    }
+    if (numCoords == 0) {
+      return {};
+    }
+
     s2Points.reserve(numCoords);
     for (size_t i = 0; i < numCoords; ++i) {
       auto coord = coordList[i];
diff --git a/src/common/geo/test/GeoFunctionTest.cpp b/src/common/geo/test/GeoFunctionTest.cpp
index d525536691e..5bfc5ba489d 100644
--- a/src/common/geo/test/GeoFunctionTest.cpp
+++ b/src/common/geo/test/GeoFunctionTest.cpp
@@ -546,43 +546,43 @@ TEST(DWithin, point2Point) {
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto point2 = Geography::fromWKT("POINT(1.0 1.0)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 0.0, true);
+    bool b = GeoFunction::dWithin(point1, point2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto point2 = Geography::fromWKT("POINT(1.0 1.0)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 0.0, false);
+    bool b = GeoFunction::dWithin(point1, point2, 0.0, true);
     EXPECT_EQ(false, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto point2 = Geography::fromWKT("POINT(1.0 1.0)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 0.1, false);
+    bool b = GeoFunction::dWithin(point1, point2, 0.1, true);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto point2 = Geography::fromWKT("POINT(1.0 1.2)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 22239.020235496788, true);
+    bool b = GeoFunction::dWithin(point1, point2, 22239.020235496788, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto point2 = Geography::fromWKT("POINT(1.0 1.2)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 22239.020235496788, false);
+    bool b = GeoFunction::dWithin(point1, point2, 22239.020235496788, true);
     EXPECT_EQ(false, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto point2 = Geography::fromWKT("POINT(1.0 1.2)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 22240, true);
+    bool b = GeoFunction::dWithin(point1, point2, 22240, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(-118.4079 33.9434)").value();
     auto point2 = Geography::fromWKT("POINT(2.5559 49.0083)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 10000, false);
+    bool b = GeoFunction::dWithin(point1, point2, 10000, true);
     EXPECT_EQ(false, b);
   }
 }
@@ -591,25 +591,25 @@ TEST(DWithin, point2LineString) {
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto line2 = Geography::fromWKT("LINESTRING(1.0 1.0, 2.0 2.0)").value();
-    bool b = GeoFunction::dWithin(point1, line2, 0.0, true);
+    bool b = GeoFunction::dWithin(point1, line2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.5 1.6)").value();
     auto line2 = Geography::fromWKT("LINESTRING(1.0 1.0, 1.5 1.6, 2.0 2.2)").value();
-    bool b = GeoFunction::dWithin(point1, line2, 0.0, true);
+    bool b = GeoFunction::dWithin(point1, line2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.5 1.6)").value();
     auto line2 = Geography::fromWKT("LINESTRING(1.0 1.0, 2.0 2.2, 1.5 1.6)").value();
-    bool b = GeoFunction::dWithin(point1, line2, 0.0, true);
+    bool b = GeoFunction::dWithin(point1, line2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.2 3.8)").value();
     auto point2 = Geography::fromWKT("LINESTRING(1.0 1.0, 1.5 1.6, 2.0 2.2)").value();
-    bool b = GeoFunction::dWithin(point1, point2, 1.0, false);
+    bool b = GeoFunction::dWithin(point1, point2, 1.0, true);
     EXPECT_EQ(false, b);
   }
 }
@@ -618,28 +618,28 @@ TEST(DWithin, point2Polygon) {
   {
     auto point1 = Geography::fromWKT("POINT(1.0 1.0)").value();
     auto polygon2 = Geography::fromWKT("POLYGON((1.0 1.0, 2.0 2.0, 0.0 2.0, 1.0 1.0))").value();
-    bool b = GeoFunction::dWithin(point1, polygon2, 0.0, true);
+    bool b = GeoFunction::dWithin(point1, polygon2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   // Point lies on the interior of the polygon
   {
     auto point1 = Geography::fromWKT("POINT(1.5 1.9)").value();
     auto polygon2 = Geography::fromWKT("POLYGON((1.0 1.0, 2.0 2.0, 0.0 2.0, 1.0 1.0))").value();
-    bool b = GeoFunction::dWithin(point1, polygon2, 0.0, true);
+    bool b = GeoFunction::dWithin(point1, polygon2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(0.5 0.5)").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
-    bool b = GeoFunction::dWithin(point1, polygon2, 0.0, true);
+    bool b = GeoFunction::dWithin(point1, polygon2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto point1 = Geography::fromWKT("POINT(1.5 1.6)").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
-    bool b = GeoFunction::dWithin(point1, polygon2, 99999, false);
+    bool b = GeoFunction::dWithin(point1, polygon2, 99999, true);
     EXPECT_EQ(true, b);
   }
   {
@@ -648,7 +648,7 @@ TEST(DWithin, point2Polygon) {
                         "POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0), (0.2 0.2, 0.2 0.4, "
                         "0.4 0.4, 0.4 0.2, 0.2 0.2))")
                         .value();
-    bool b = GeoFunction::dWithin(point1, polygon2, 10000, true);
+    bool b = GeoFunction::dWithin(point1, polygon2, 10000, false);
     EXPECT_EQ(false, b);
   }
 }
@@ -657,31 +657,31 @@ TEST(DWithin, lineString2LineString) {
   {
     auto line1 = Geography::fromWKT("LINESTRING(2.0 2.0, 3.0 3.0)").value();
     auto line2 = Geography::fromWKT("LINESTRING(2.0 2.0, 3.0 3.0)").value();
-    bool b = GeoFunction::dWithin(line1, line2, 0.0, true);
+    bool b = GeoFunction::dWithin(line1, line2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto line1 = Geography::fromWKT("LINESTRING(1.0 1.0, 1.5 1.5, 2.0 2.0)").value();
     auto line2 = Geography::fromWKT("LINESTRING(2.0 2.0, 3.0 3.0)").value();
-    bool b = GeoFunction::dWithin(line1, line2, 0.0, true);
+    bool b = GeoFunction::dWithin(line1, line2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto line1 = Geography::fromWKT("LINESTRING(0.0 0.0, 1.0 1.0)").value();
     auto line2 = Geography::fromWKT("LINESTRING(5.0 5.0, 6.0 6.0)").value();
-    bool b = GeoFunction::dWithin(line1, line2, 628519, true);
+    bool b = GeoFunction::dWithin(line1, line2, 628519, false);
     EXPECT_EQ(false, b);
   }
   {
     auto line1 = Geography::fromWKT("LINESTRING(1.0 1.0, 1.5 1.5, 2.0 2.0)").value();
     auto line2 = Geography::fromWKT("LINESTRING(1.0 2.0, 2.0 1.0)").value();
-    bool b = GeoFunction::dWithin(line1, line2, 1.0, false);
+    bool b = GeoFunction::dWithin(line1, line2, 1.0, true);
     EXPECT_EQ(true, b);
   }
   {
     auto line1 = Geography::fromWKT("LINESTRING(1.0 1.0, 1.5 1.5, 2.0 2.0)").value();
     auto line2 = Geography::fromWKT("LINESTRING(1.0 2.0, 1.0 4.0)").value();
-    bool b = GeoFunction::dWithin(line1, line2, 78609, false);
+    bool b = GeoFunction::dWithin(line1, line2, 78609, true);
     EXPECT_EQ(true, b);
   }
 }
@@ -691,21 +691,21 @@ TEST(DWithin, lineString2Polygon) {
     auto line1 = Geography::fromWKT("LINESTRING(1.0 1.0, 1.5 1.5, 2.0 2.0)").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
-    bool b = GeoFunction::dWithin(line1, polygon2, 0.0, true);
+    bool b = GeoFunction::dWithin(line1, polygon2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
     auto line1 = Geography::fromWKT("LINESTRING(0.2 0.2, 0.4 0.4)").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
-    bool b = GeoFunction::dWithin(line1, polygon2, 0.0, false);
+    bool b = GeoFunction::dWithin(line1, polygon2, 0.0, true);
     EXPECT_EQ(false, b);
   }
   {
     auto line1 = Geography::fromWKT("LINESTRING(-0.5 0.5, 0.5 0.5)").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
-    bool b = GeoFunction::dWithin(line1, polygon2, 9999, false);
+    bool b = GeoFunction::dWithin(line1, polygon2, 9999, true);
     EXPECT_EQ(true, b);
   }
   {
@@ -714,7 +714,7 @@ TEST(DWithin, lineString2Polygon) {
                         "POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0), (0.2 0.2, 0.2 0.4, "
                         "0.4 0.4, 0.4 0.2, 0.2 0.2))")
                         .value();
-    bool b = GeoFunction::dWithin(line1, polygon2, 5555, true);
+    bool b = GeoFunction::dWithin(line1, polygon2, 5555, false);
     EXPECT_EQ(false, b);
   }
   {
@@ -723,7 +723,7 @@ TEST(DWithin, lineString2Polygon) {
                         "POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0), (0.2 0.2, 0.2 0.4, "
                         "0.4 0.4, 0.4 0.2, 0.2 0.2))")
                         .value();
-    bool b = GeoFunction::dWithin(line1, polygon2, 0.1, false);
+    bool b = GeoFunction::dWithin(line1, polygon2, 0.1, true);
     EXPECT_EQ(true, b);
   }
   {
@@ -732,7 +732,7 @@ TEST(DWithin, lineString2Polygon) {
                         "POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0), (0.2 0.2, 0.2 0.4, "
                         "0.4 0.4, 0.4 0.2, 0.2 0.2))")
                         .value();
-    bool b = GeoFunction::dWithin(line1, polygon2, 628520, false);
+    bool b = GeoFunction::dWithin(line1, polygon2, 628520, true);
     EXPECT_EQ(true, b);
   }
 }
@@ -743,7 +743,7 @@ TEST(DWithin, polygon2Polygon) {
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
-    bool b = GeoFunction::dWithin(polygon1, polygon2, 0.0, true);
+    bool b = GeoFunction::dWithin(polygon1, polygon2, 0.0, false);
     EXPECT_EQ(true, b);
   }
   {
@@ -751,7 +751,7 @@ TEST(DWithin, polygon2Polygon) {
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((0.2 0.2, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.2 0.2))").value();
-    bool b = GeoFunction::dWithin(polygon1, polygon2, 1.1, false);
+    bool b = GeoFunction::dWithin(polygon1, polygon2, 1.1, true);
     EXPECT_EQ(true, b);
   }
   {
@@ -759,7 +759,7 @@ TEST(DWithin, polygon2Polygon) {
         Geography::fromWKT("POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0))").value();
     auto polygon2 =
         Geography::fromWKT("POLYGON((3.0 3.0, 4.0 3.0, 4.0 4.0, 3.0 4.0, 3.0 3.0))").value();
-    bool b = GeoFunction::dWithin(polygon1, polygon2, 314402, true);
+    bool b = GeoFunction::dWithin(polygon1, polygon2, 314402, false);
     EXPECT_EQ(false, b);
   }
   {
@@ -770,7 +770,7 @@ TEST(DWithin, polygon2Polygon) {
                         "POLYGON((0.0 0.0, 1.0 0.0, 1.0 1.0, 0.0 1.0, 0.0 0.0), (0.2 0.2, 0.2 0.4, "
                         "0.4 0.4, 0.4 0.2, 0.2 0.2))")
                         .value();
-    bool b = GeoFunction::dWithin(polygon1, polygon2, 5560, false);
+    bool b = GeoFunction::dWithin(polygon1, polygon2, 5560, true);
     EXPECT_EQ(true, b);
   }
   {
@@ -779,7 +779,7 @@ TEST(DWithin, polygon2Polygon) {
     auto polygon2 =
         Geography::fromWKT("POLYGON((-8.0 -8.0, -4.0 -8.0, -4.0 -4.0, -8.0 -4.0, -8.0 -8.0))")
             .value();
-    bool b = GeoFunction::dWithin(polygon1, polygon2, 628750, false);
+    bool b = GeoFunction::dWithin(polygon1, polygon2, 628750, true);
     EXPECT_EQ(false, b);
   }
 }
diff --git a/src/common/meta/NebulaSchemaProvider.cpp b/src/common/meta/NebulaSchemaProvider.cpp
index dc5aab4a2c5..a9bf2f5e66f 100644
--- a/src/common/meta/NebulaSchemaProvider.cpp
+++ b/src/common/meta/NebulaSchemaProvider.cpp
@@ -91,7 +91,8 @@ void NebulaSchemaProvider::addField(folly::StringPiece name,
                                     cpp2::PropertyType type,
                                     size_t fixedStrLen,
                                     bool nullable,
-                                    Expression* defaultValue) {
+                                    Expression* defaultValue,
+                                    cpp2::GeoShape geoShape) {
   size_t size = fieldSize(type, fixedStrLen);
 
   size_t offset = 0;
@@ -112,7 +113,8 @@ void NebulaSchemaProvider::addField(folly::StringPiece name,
                        defaultValue,
                        size,
                        offset,
-                       nullFlagPos);
+                       nullFlagPos,
+                       geoShape);
   fieldNameIndex_.emplace(name.toString(), static_cast<int64_t>(fields_.size() - 1));
 }
 
diff --git a/src/common/meta/NebulaSchemaProvider.h b/src/common/meta/NebulaSchemaProvider.h
index 0b185c7ed3a..31a381a4a83 100644
--- a/src/common/meta/NebulaSchemaProvider.h
+++ b/src/common/meta/NebulaSchemaProvider.h
@@ -29,7 +29,8 @@ class NebulaSchemaProvider : public SchemaProviderIf {
                 Expression* defaultValue,
                 size_t size,
                 size_t offset,
-                size_t nullFlagPos)
+                size_t nullFlagPos,
+                cpp2::GeoShape geoShape)
         : name_(std::move(name)),
           type_(std::move(type)),
           nullable_(nullable),
@@ -37,7 +38,8 @@ class NebulaSchemaProvider : public SchemaProviderIf {
           defaultValue_(defaultValue),
           size_(size),
           offset_(offset),
-          nullFlagPos_(nullFlagPos) {}
+          nullFlagPos_(nullFlagPos),
+          geoShape_(geoShape) {}
 
     const char* name() const override { return name_.c_str(); }
 
@@ -58,6 +60,8 @@ class NebulaSchemaProvider : public SchemaProviderIf {
       return nullFlagPos_;
     }
 
+    cpp2::GeoShape geoShape() const override { return geoShape_; }
+
    private:
     std::string name_;
     cpp2::PropertyType type_;
@@ -67,6 +71,7 @@ class NebulaSchemaProvider : public SchemaProviderIf {
     size_t size_;
     size_t offset_;
     size_t nullFlagPos_;
+    cpp2::GeoShape geoShape_;
   };
 
  public:
@@ -93,7 +98,8 @@ class NebulaSchemaProvider : public SchemaProviderIf {
                 cpp2::PropertyType type,
                 size_t fixedStrLen = 0,
                 bool nullable = false,
-                Expression* defaultValue = nullptr);
+                Expression* defaultValue = nullptr,
+                cpp2::GeoShape geoShape = cpp2::GeoShape::ANY);
 
   static std::size_t fieldSize(cpp2::PropertyType type, std::size_t fixedStrLimit);
 
diff --git a/src/common/meta/SchemaProviderIf.h b/src/common/meta/SchemaProviderIf.h
index b880c7f01cd..f643aab28cb 100644
--- a/src/common/meta/SchemaProviderIf.h
+++ b/src/common/meta/SchemaProviderIf.h
@@ -40,6 +40,10 @@ class SchemaProviderIf {
     // In v2, if the field is nullable, it returns the position of
     // the null flag bit, otherwise, it returns 0
     virtual size_t nullFlagPos() const = 0;
+    // In v1, this always returns cpp2::GeoShape::ANY
+    // In v2, if the field type is cpp2::PropertyType::Geography,
+    // it returns the specified geo shape type
+    virtual cpp2::GeoShape geoShape() const = 0;
   };
 
   // Inherited classes do not need to implement the Iterator
diff --git a/src/common/utils/IndexKeyUtils.h b/src/common/utils/IndexKeyUtils.h
index 23d699abce8..aaf2fe1cfd1 100644
--- a/src/common/utils/IndexKeyUtils.h
+++ b/src/common/utils/IndexKeyUtils.h
@@ -10,6 +10,7 @@
 #include "codec/RowReader.h"
 #include "common/base/Base.h"
 #include "common/base/StatusOr.h"
+#include "common/geo/GeoIndex.h"
 #include "common/utils/Types.h"
 #include "interface/gen-cpp2/meta_types.h"
 
@@ -138,8 +139,7 @@ class IndexKeyUtils final {
         return encodeDateTime(v.getDateTime());
       }
       case Value::Type::GEOGRAPHY: {
-        // TODO(jie)
-        return "";
+        return encodeGeography(v.getGeography());
       }
       default:
         LOG(ERROR) << "Unsupported default value type";
@@ -299,6 +299,20 @@ class IndexKeyUtils final {
     return buf;
   }
 
+  static std::string encodeGeography(const nebula::Geography& gg) {
+    // get params from index meta
+    geo::RegionCoverParams
+        rc;  // TODO(jie): Get index params to build to construct RegionCoverParams
+    geo::GeoIndex geoIndex(rc, false);  // TODO(jie): Get schema meta to know if it's point only
+    auto cellIds = geoIndex.indexCells(gg);
+    std::vector<std::string> bufs;
+    for (auto cellId : cellIds) {
+      bufs.emplace_back(encodeUint64(cellId));
+    }
+
+    return bufs[0];  // just support index point here.
+  }
+
   static nebula::DateTime decodeDateTime(const folly::StringPiece& raw) {
     int16_t year = *reinterpret_cast<const int16_t*>(raw.data());
     int8_t month = *reinterpret_cast<const int8_t*>(raw.data() + sizeof(int16_t));
diff --git a/src/common/utils/test/CMakeLists.txt b/src/common/utils/test/CMakeLists.txt
index 7e9d59739ff..d02df8575dc 100644
--- a/src/common/utils/test/CMakeLists.txt
+++ b/src/common/utils/test/CMakeLists.txt
@@ -11,6 +11,7 @@ nebula_add_test(
         $<TARGET_OBJECTS:wkt_wkb_io_obj>
         $<TARGET_OBJECTS:meta_thrift_obj>
         $<TARGET_OBJECTS:common_thrift_obj>
+        $<TARGET_OBJECTS:geo_index_obj>
     LIBRARIES
         gtest
         ${THRIFT_LIBRARIES}
@@ -29,6 +30,7 @@ nebula_add_test(
         $<TARGET_OBJECTS:wkt_wkb_io_obj>
         $<TARGET_OBJECTS:meta_thrift_obj>
         $<TARGET_OBJECTS:common_thrift_obj>
+        $<TARGET_OBJECTS:geo_index_obj>
     LIBRARIES
         gtest
         ${THRIFT_LIBRARIES}
@@ -47,6 +49,7 @@ nebula_add_test(
         $<TARGET_OBJECTS:wkt_wkb_io_obj>
         $<TARGET_OBJECTS:meta_thrift_obj>
         $<TARGET_OBJECTS:common_thrift_obj>
+        $<TARGET_OBJECTS:geo_index_obj>
     LIBRARIES
         gtest
         ${THRIFT_LIBRARIES}
diff --git a/src/daemons/CMakeLists.txt b/src/daemons/CMakeLists.txt
index 13eaa0e673e..05aef32acd0 100644
--- a/src/daemons/CMakeLists.txt
+++ b/src/daemons/CMakeLists.txt
@@ -45,6 +45,7 @@ set(storage_meta_deps
     $<TARGET_OBJECTS:raftex_thrift_obj>
     $<TARGET_OBJECTS:hdfs_helper_obj>
     $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
+    $<TARGET_OBJECTS:geo_index_obj>
 )
 
 nebula_add_executable(
diff --git a/src/graph/util/SchemaUtil.cpp b/src/graph/util/SchemaUtil.cpp
index ccbcb9ca028..c638efa5365 100644
--- a/src/graph/util/SchemaUtil.cpp
+++ b/src/graph/util/SchemaUtil.cpp
@@ -67,7 +67,8 @@ std::shared_ptr<const meta::NebulaSchemaProvider> SchemaUtil::generateSchemaProv
                         col.get_type().get_type(),
                         col.type.type_length_ref().value_or(0),
                         col.nullable_ref().value_or(false),
-                        hasDef ? defaultValueExpr : nullptr);
+                        hasDef ? defaultValueExpr : nullptr,
+                        col.type.geo_shape_ref().value_or(meta::cpp2::GeoShape::ANY));
   }
   return schemaPtr;
 }
@@ -259,6 +260,13 @@ std::string SchemaUtil::typeToString(const meta::cpp2::ColumnTypeDef &col) {
   auto type = apache::thrift::util::enumNameSafe(col.get_type());
   if (col.get_type() == meta::cpp2::PropertyType::FIXED_STRING) {
     return folly::stringPrintf("%s(%d)", type.c_str(), *col.get_type_length());
+  } else if (col.get_type() == meta::cpp2::PropertyType::GEOGRAPHY) {
+    auto geoShape = *col.get_geo_shape();
+    if (geoShape == meta::cpp2::GeoShape::ANY) {
+      return folly::stringPrintf("%s", type.c_str());
+    }
+    return folly::stringPrintf(
+        "%s(%s)", type.c_str(), apache::thrift::util::enumNameSafe(geoShape).c_str());
   }
   return type;
 }
diff --git a/src/graph/validator/LookupValidator.cpp b/src/graph/validator/LookupValidator.cpp
index 8d4f421d6ba..747a1567a24 100644
--- a/src/graph/validator/LookupValidator.cpp
+++ b/src/graph/validator/LookupValidator.cpp
@@ -102,6 +102,32 @@ void LookupValidator::extractExprProps() {
   lookupCtx_->idxReturnCols = std::move(idxReturnCols_);
 }
 
+bool LookupValidator::isGeoIndexAcceleratedPredicate(const Expression* expr) const {
+  static std::unordered_set<std::string> geoIndexAcceleratedPredicates{
+      "st_intersects", "st_covers", "st_coveredby", "st_dwithin"};
+
+  if (expr->isRelExpr()) {
+    auto* relExpr = static_cast<const RelationalExpression*>(expr);
+    auto isSt_Distance = [](const Expression* e) {
+      if (e->kind() == Expression::Kind::kFunctionCall) {
+        auto* funcExpr = static_cast<const FunctionCallExpression*>(e);
+        return boost::algorithm::iequals(funcExpr->name(), "st_distance");
+      }
+      return false;
+    };
+    return isSt_Distance(relExpr->left()) || isSt_Distance(relExpr->right());
+  } else if (expr->kind() == Expression::Kind::kFunctionCall) {
+    auto* funcExpr = static_cast<const FunctionCallExpression*>(expr);
+    std::string funcName = funcExpr->name();
+    folly::toLowerAscii(funcName);
+    if (geoIndexAcceleratedPredicates.count(funcName) != 0) {
+      return true;
+    }
+  }
+
+  return false;
+}
+
 Status LookupValidator::validateYieldEdge() {
   auto yield = sentence()->yieldClause();
   auto yieldExpr = lookupCtx_->yieldExpr;
@@ -260,7 +286,10 @@ StatusOr<Expression*> LookupValidator::handleLogicalExprOperands(LogicalExpressi
 
 StatusOr<Expression*> LookupValidator::checkFilter(Expression* expr) {
   // TODO: Support IN expression push down
-  if (expr->isRelExpr()) {
+  if (isGeoIndexAcceleratedPredicate(expr)) {
+    NG_RETURN_IF_ERROR(checkGeoPredicate(expr));
+    return rewriteGeoPredicate(expr);
+  } else if (expr->isRelExpr()) {
     // Only starts with can be pushed down as a range scan, so forbid other string-related relExpr
     if (expr->kind() == ExprKind::kRelREG || expr->kind() == ExprKind::kContains ||
         expr->kind() == ExprKind::kNotContains || expr->kind() == ExprKind::kEndsWith ||
@@ -299,9 +328,47 @@ Status LookupValidator::checkRelExpr(RelationalExpression* expr) {
   if (left->kind() == ExprKind::kLabelAttribute || right->kind() == ExprKind::kLabelAttribute) {
     return Status::OK();
   }
+
   return Status::SemanticError("Expression %s not supported yet", expr->toString().c_str());
 }
 
+Status LookupValidator::checkGeoPredicate(const Expression* expr) const {
+  auto checkFunc = [](const FunctionCallExpression* funcExpr) -> Status {
+    if (funcExpr->args()->numArgs() < 2) {
+      return Status::SemanticError("Expression %s has not enough arguments",
+                                   funcExpr->toString().c_str());
+    }
+    auto* first = funcExpr->args()->args()[0];
+    auto* second = funcExpr->args()->args()[1];
+
+    if (first->kind() == ExprKind::kLabelAttribute && second->kind() == ExprKind::kLabelAttribute) {
+      return Status::SemanticError("Expression %s not supported yet", funcExpr->toString().c_str());
+    }
+    if (first->kind() == ExprKind::kLabelAttribute || second->kind() == ExprKind::kLabelAttribute) {
+      return Status::OK();
+    }
+
+    return Status::SemanticError("Expression %s not supported yet", funcExpr->toString().c_str());
+  };
+  if (expr->isRelExpr()) {
+    auto* relExpr = static_cast<const RelationalExpression*>(expr);
+    auto* left = relExpr->left();
+    auto* right = relExpr->right();
+    if (left->kind() == Expression::Kind::kFunctionCall) {
+      NG_RETURN_IF_ERROR(checkFunc(static_cast<const FunctionCallExpression*>(left)));
+    }
+    if (right->kind() == Expression::Kind::kFunctionCall) {
+      NG_RETURN_IF_ERROR(checkFunc(static_cast<const FunctionCallExpression*>(right)));
+    }
+  } else if (expr->kind() == Expression::Kind::kFunctionCall) {
+    NG_RETURN_IF_ERROR(checkFunc(static_cast<const FunctionCallExpression*>(expr)));
+  } else {
+    return Status::SemanticError("Expression %s not supported yet", expr->toString().c_str());
+  }
+
+  return Status::OK();
+}
+
 StatusOr<Expression*> LookupValidator::rewriteRelExpr(RelationalExpression* expr) {
   // swap LHS and RHS of relExpr if LabelAttributeExpr in on the right,
   // so that LabelAttributeExpr is always on the left
@@ -336,6 +403,73 @@ StatusOr<Expression*> LookupValidator::rewriteRelExpr(RelationalExpression* expr
   return expr;
 }
 
+StatusOr<Expression*> LookupValidator::rewriteGeoPredicate(Expression* expr) {
+  // swap LHS and RHS of relExpr if LabelAttributeExpr in on the right,
+  // so that LabelAttributeExpr is always on the left
+  if (expr->isRelExpr()) {
+    if (static_cast<RelationalExpression*>(expr)->right()->kind() ==
+        Expression::Kind::kFunctionCall) {
+      expr = reverseGeoPredicate(expr);
+    }
+    auto* relExpr = static_cast<RelationalExpression*>(expr);
+    auto* left = relExpr->left();
+    auto* right = relExpr->right();
+    DCHECK(left->kind() == Expression::Kind::kFunctionCall);
+    auto* funcExpr = static_cast<FunctionCallExpression*>(left);
+    DCHECK_EQ(boost::to_lower_copy(funcExpr->name()), "st_distance");
+    // `ST_Distance(g1, g1) < distanceInMeters` is equivalent to `ST_DWithin(g1, g2,
+    // distanceInMeters, true), `ST_Distance(g1, g1) <= distanceInMeters` is equivalent to
+    // `ST_DWithin(g1, g2, distanceInMeters, false)
+    if (relExpr->kind() == Expression::Kind::kRelLT ||
+        relExpr->kind() == Expression::Kind::kRelLE) {
+      auto* newArgList = ArgumentList::make(qctx_->objPool(), 3);
+      if (funcExpr->args()->numArgs() != 2) {
+        return Status::SemanticError("Expression %s has wrong number arguments",
+                                     funcExpr->toString().c_str());
+      }
+      newArgList->addArgument(funcExpr->args()->args()[0]->clone());
+      newArgList->addArgument(funcExpr->args()->args()[1]->clone());
+      newArgList->addArgument(right->clone());  // distanceInMeters
+      bool exclusive = relExpr->kind() == Expression::Kind::kRelLT;
+      newArgList->addArgument(ConstantExpression::make(qctx_->objPool(), exclusive));
+      expr = FunctionCallExpression::make(qctx_->objPool(), "st_dwithin", newArgList);
+    } else {
+      return Status::SemanticError("Expression %s not supported yet", expr->toString().c_str());
+    }
+  }
+
+  DCHECK(expr->kind() == Expression::Kind::kFunctionCall);
+  if (static_cast<FunctionCallExpression*>(expr)->args()->args()[1]->kind() ==
+      ExprKind::kLabelAttribute) {
+    expr = reverseGeoPredicate(expr);
+  }
+
+  auto* geoFuncExpr = static_cast<FunctionCallExpression*>(expr);
+  auto* first = geoFuncExpr->args()->args()[0];
+  auto* la = static_cast<LabelAttributeExpression*>(first);
+  if (la->left()->name() != sentence()->from()) {
+    return Status::SemanticError("Schema name error: %s", la->left()->name().c_str());
+  }
+
+  // fold constant expression
+  auto foldRes = ExpressionUtils::foldConstantExpr(expr);
+  NG_RETURN_IF_ERROR(foldRes);
+  expr = static_cast<FunctionCallExpression*>(foldRes.value());
+  DCHECK_EQ(geoFuncExpr->args()->args()[0]->kind(), ExprKind::kLabelAttribute);
+
+  // Check schema and value type
+  std::string prop = la->right()->value().getStr();
+  auto c = checkConstExpr(geoFuncExpr->args()->args()[1], prop, Expression::Kind::kFunctionCall);
+  NG_RETURN_IF_ERROR(c);
+  geoFuncExpr->args()->setArg(1, std::move(c).value());
+
+  // rewrite PropertyExpression
+  auto propExpr = lookupCtx_->isEdge ? ExpressionUtils::rewriteLabelAttr2EdgeProp(la)
+                                     : ExpressionUtils::rewriteLabelAttr2TagProp(la);
+  geoFuncExpr->args()->setArg(0, propExpr);
+  return geoFuncExpr;
+}
+
 StatusOr<Expression*> LookupValidator::checkConstExpr(Expression* expr,
                                                       const std::string& prop,
                                                       const ExprKind kind) {
@@ -437,6 +571,35 @@ Expression* LookupValidator::reverseRelKind(RelationalExpression* expr) {
   return RelationalExpression::makeKind(pool, reversedKind, right->clone(), left->clone());
 }
 
+Expression* LookupValidator::reverseGeoPredicate(Expression* expr) {
+  if (expr->isRelExpr()) {
+    auto* relExpr = static_cast<RelationalExpression*>(expr);
+    return reverseRelKind(relExpr);
+  } else {
+    DCHECK(expr->kind() == Expression::Kind::kFunctionCall);
+    auto* funcExpr = static_cast<const FunctionCallExpression*>(expr);
+    auto name = funcExpr->name();
+    folly::toLowerAscii(name);
+    auto newName = name;
+
+    if (name == "st_covers") {
+      newName = "st_coveredby";
+    } else if (name == "st_coveredby") {
+      newName = "st_covers";
+    } else if (name == "st_intersects") {
+    } else if (name == "st_dwithin") {
+    }
+
+    auto* newArgList = ArgumentList::make(qctx_->objPool(), funcExpr->args()->numArgs());
+    for (auto& arg : funcExpr->args()->args()) {
+      newArgList->addArgument(arg->clone());
+    }
+    newArgList->setArg(0, funcExpr->args()->args()[1]);
+    newArgList->setArg(1, funcExpr->args()->args()[0]);
+    return FunctionCallExpression::make(qctx_->objPool(), newName, newArgList);
+  }
+}
+
 Status LookupValidator::getSchemaProvider(shared_ptr<const NebulaSchemaProvider>* provider) const {
   auto from = sentence()->from();
   auto schemaMgr = qctx_->schemaMng();
diff --git a/src/graph/validator/LookupValidator.h b/src/graph/validator/LookupValidator.h
index a0a1fd859d0..2b4e906ee65 100644
--- a/src/graph/validator/LookupValidator.h
+++ b/src/graph/validator/LookupValidator.h
@@ -39,12 +39,15 @@ class LookupValidator final : public Validator {
 
   StatusOr<Expression*> checkFilter(Expression* expr);
   Status checkRelExpr(RelationalExpression* expr);
+  Status checkGeoPredicate(const Expression* expr) const;
   StatusOr<std::string> checkTSExpr(Expression* expr);
   StatusOr<Expression*> checkConstExpr(Expression* expr,
                                        const std::string& prop,
                                        const Expression::Kind kind);
   StatusOr<Expression*> rewriteRelExpr(RelationalExpression* expr);
+  StatusOr<Expression*> rewriteGeoPredicate(Expression* expr);
   Expression* reverseRelKind(RelationalExpression* expr);
+  Expression* reverseGeoPredicate(Expression* expr);
 
   const LookupSentence* sentence() const;
   int32_t schemaId() const;
@@ -56,6 +59,7 @@ class LookupValidator final : public Validator {
   StatusOr<Expression*> handleLogicalExprOperands(LogicalExpression* lExpr);
   Status extractSchemaProp();
   void extractExprProps();
+  bool isGeoIndexAcceleratedPredicate(const Expression* expr) const;
 
   std::unique_ptr<LookupContext> lookupCtx_;
   std::vector<nebula::plugin::HttpClient> tsClients_;
diff --git a/src/graph/validator/MaintainValidator.cpp b/src/graph/validator/MaintainValidator.cpp
index ca04be664ee..345b31d1039 100644
--- a/src/graph/validator/MaintainValidator.cpp
+++ b/src/graph/validator/MaintainValidator.cpp
@@ -37,6 +37,8 @@ Status SchemaValidator::validateColumns(const std::vector<ColumnSpecification *>
     column.type.set_type(type);
     if (meta::cpp2::PropertyType::FIXED_STRING == type) {
       column.type.set_type_length(spec->typeLen());
+    } else if (meta::cpp2::PropertyType::GEOGRAPHY == type) {
+      column.type.set_geo_shape(spec->geoShape());
     }
     for (const auto &property : spec->properties()->properties()) {
       if (property->isNullable()) {
diff --git a/src/kvstore/test/CMakeLists.txt b/src/kvstore/test/CMakeLists.txt
index 6f3db67d2cd..aa039b23854 100644
--- a/src/kvstore/test/CMakeLists.txt
+++ b/src/kvstore/test/CMakeLists.txt
@@ -33,6 +33,7 @@ set(KVSTORE_TEST_LIBS
     $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
     $<TARGET_OBJECTS:version_obj>
     $<TARGET_OBJECTS:ssl_obj>
+    $<TARGET_OBJECTS:geo_index_obj>
 )
 
 nebula_add_test(
diff --git a/src/meta/CMakeLists.txt b/src/meta/CMakeLists.txt
index ea9f436d611..90c4835dd0b 100644
--- a/src/meta/CMakeLists.txt
+++ b/src/meta/CMakeLists.txt
@@ -162,6 +162,7 @@ set(meta_test_deps
     $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
     $<TARGET_OBJECTS:version_obj>
     $<TARGET_OBJECTS:ssl_obj>
+    $<TARGET_OBJECTS:geo_index_obj>
 )
 
 nebula_add_subdirectory(http)
diff --git a/src/parser/test/ParserTest.cpp b/src/parser/test/ParserTest.cpp
index 8ccc3490942..4d647db9960 100644
--- a/src/parser/test/ParserTest.cpp
+++ b/src/parser/test/ParserTest.cpp
@@ -91,6 +91,27 @@ TEST_F(ParserTest, TestSchemaCreation) {
     auto result = parse(query);
     ASSERT_TRUE(result.ok()) << result.status();
   }
+  // Geo spatial
+  {
+    std::string query = "CREATE TAG any_shape(geo geography)";
+    auto result = parse(query);
+    ASSERT_TRUE(result.ok()) << result.status();
+  }
+  {
+    std::string query = "CREATE TAG any_shape(geo geography(point))";
+    auto result = parse(query);
+    ASSERT_TRUE(result.ok()) << result.status();
+  }
+  {
+    std::string query = "CREATE TAG any_shape(geo geography(linestring))";
+    auto result = parse(query);
+    ASSERT_TRUE(result.ok()) << result.status();
+  }
+  {
+    std::string query = "CREATE TAG any_shape(geo geography(polygon))";
+    auto result = parse(query);
+    ASSERT_TRUE(result.ok()) << result.status();
+  }
 }
 
 TEST_F(ParserTest, Go) {
diff --git a/src/storage/index/LookupBaseProcessor-inl.h b/src/storage/index/LookupBaseProcessor-inl.h
index 5fb46d38a83..6a280f078bb 100644
--- a/src/storage/index/LookupBaseProcessor-inl.h
+++ b/src/storage/index/LookupBaseProcessor-inl.h
@@ -210,7 +210,11 @@ StatusOr<StoragePlan<IndexID>> LookupBaseProcessor<REQ, RESP>::buildPlan(
       auto it = std::find_if(fields.begin(), fields.end(), [&yieldCol](const auto& columnDef) {
         return yieldCol == columnDef.get_name();
       });
-      if (it == fields.end()) {
+      if (it == fields.end() ||
+          it->get_type().get_type() ==
+              nebula::meta::cpp2::PropertyType::GEOGRAPHY) {  // geography index just stores
+                                                              // S2CellId, so must read the
+                                                              // original geo data.
         needData = true;
         break;
       }
diff --git a/src/storage/test/CMakeLists.txt b/src/storage/test/CMakeLists.txt
index 33dee520e63..9ad3b3e7529 100644
--- a/src/storage/test/CMakeLists.txt
+++ b/src/storage/test/CMakeLists.txt
@@ -49,6 +49,7 @@ set(storage_test_deps
     $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
     $<TARGET_OBJECTS:version_obj>
     $<TARGET_OBJECTS:ssl_obj>
+    $<TARGET_OBJECTS:geo_index_obj>
 )
 
 nebula_add_test(
diff --git a/src/tools/db-dump/CMakeLists.txt b/src/tools/db-dump/CMakeLists.txt
index d48e0f73575..893bd53e494 100644
--- a/src/tools/db-dump/CMakeLists.txt
+++ b/src/tools/db-dump/CMakeLists.txt
@@ -45,6 +45,7 @@ set(tools_test_deps
     $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
     $<TARGET_OBJECTS:version_obj>
     $<TARGET_OBJECTS:ssl_obj>
+    $<TARGET_OBJECTS:geo_index_obj>
 )
 
 nebula_add_executable(
diff --git a/src/tools/db-upgrade/CMakeLists.txt b/src/tools/db-upgrade/CMakeLists.txt
index 7b3670d20ee..4aa3f68be00 100644
--- a/src/tools/db-upgrade/CMakeLists.txt
+++ b/src/tools/db-upgrade/CMakeLists.txt
@@ -53,6 +53,7 @@ nebula_add_executable(
         $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
         $<TARGET_OBJECTS:version_obj>
         $<TARGET_OBJECTS:ssl_obj>
+        $<TARGET_OBJECTS:geo_index_obj>
     LIBRARIES
         ${ROCKSDB_LIBRARIES}
         ${THRIFT_LIBRARIES}
diff --git a/src/tools/meta-dump/CMakeLists.txt b/src/tools/meta-dump/CMakeLists.txt
index 663e4686366..46c508f3b1f 100644
--- a/src/tools/meta-dump/CMakeLists.txt
+++ b/src/tools/meta-dump/CMakeLists.txt
@@ -50,6 +50,7 @@ nebula_add_executable(
         $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
         $<TARGET_OBJECTS:version_obj>
         $<TARGET_OBJECTS:ssl_obj>
+        $<TARGET_OBJECTS:geo_index_obj>
     LIBRARIES
         ${ROCKSDB_LIBRARIES}
         ${THRIFT_LIBRARIES}
diff --git a/src/tools/simple-kv-verify/CMakeLists.txt b/src/tools/simple-kv-verify/CMakeLists.txt
index 75c8b9cfb88..17a2592fe13 100644
--- a/src/tools/simple-kv-verify/CMakeLists.txt
+++ b/src/tools/simple-kv-verify/CMakeLists.txt
@@ -44,6 +44,7 @@ nebula_add_executable(
         $<TARGET_OBJECTS:wkt_wkb_io_obj>
         $<TARGET_OBJECTS:agg_function_manager_obj>
         $<TARGET_OBJECTS:version_obj>
+        $<TARGET_OBJECTS:geo_index_obj>
     LIBRARIES
         ${ROCKSDB_LIBRARIES}
         ${THRIFT_LIBRARIES}
diff --git a/src/tools/storage-perf/CMakeLists.txt b/src/tools/storage-perf/CMakeLists.txt
index 5a4400210d2..e278053ecb2 100644
--- a/src/tools/storage-perf/CMakeLists.txt
+++ b/src/tools/storage-perf/CMakeLists.txt
@@ -45,6 +45,7 @@ set(perf_test_deps
     $<TARGET_OBJECTS:ft_es_storage_adapter_obj>
     $<TARGET_OBJECTS:version_obj>
     $<TARGET_OBJECTS:ssl_obj>
+    $<TARGET_OBJECTS:geo_index_obj>
 )
 
 nebula_add_executable(
diff --git a/tests/tck/features/geo/GeoBase.feature b/tests/tck/features/geo/GeoBase.feature
new file mode 100644
index 00000000000..c6b30b0752c
--- /dev/null
+++ b/tests/tck/features/geo/GeoBase.feature
@@ -0,0 +1,526 @@
+# Copyright (c) 2020 vesoft inc. All rights reserved.
+#
+# This source code is licensed under Apache 2.0 License,
+# attached with Common Clause Condition 1.0, found in the LICENSES directory.
+Feature: Geo base
+
+  Background:
+    Given an empty graph
+    And create a space with following options:
+      | partition_num  | 9                |
+      | replica_factor | 1                |
+      | vid_type       | FIXED_STRING(30) |
+      | charset        | utf8             |
+      | collate        | utf8_bin         |
+    And having executed:
+      """
+      CREATE TAG any_shape(geo geography);
+      CREATE TAG only_point(geo geography(point));
+      CREATE TAG only_linestring(geo geography(linestring));
+      CREATE TAG only_polygon(geo geography(polygon));
+      CREATE EDGE any_shape_edge(geo geography);
+      """
+    And wait 3 seconds
+
+  Scenario: desc and show create geo schema
+    # Desc geo schema
+    When executing query:
+      """
+      DESC TAG any_shape;
+      """
+    Then the result should be, in any order:
+      | Field | Type        | Null  | Default | Comment |
+      | "geo" | "geography" | "YES" | EMPTY   | EMPTY   |
+    When executing query:
+      """
+      DESC TAG only_point;
+      """
+    Then the result should be, in any order:
+      | Field | Type               | Null  | Default | Comment |
+      | "geo" | "geography(point)" | "YES" | EMPTY   | EMPTY   |
+    When executing query:
+      """
+      DESC TAG only_linestring;
+      """
+    Then the result should be, in any order:
+      | Field | Type                    | Null  | Default | Comment |
+      | "geo" | "geography(linestring)" | "YES" | EMPTY   | EMPTY   |
+    When executing query:
+      """
+      DESC TAG only_polygon;
+      """
+    Then the result should be, in any order:
+      | Field | Type                 | Null  | Default | Comment |
+      | "geo" | "geography(polygon)" | "YES" | EMPTY   | EMPTY   |
+    When executing query:
+      """
+      DESC EDGE any_shape_edge;
+      """
+    Then the result should be, in any order:
+      | Field | Type        | Null  | Default | Comment |
+      | "geo" | "geography" | "YES" | EMPTY   | EMPTY   |
+    # Show create geo schema
+    When executing query:
+      """
+      SHOW CREATE TAG only_point;
+      """
+    Then the result should be, in any order:
+      | Tag          | Create Tag                                                                                  |
+      | "only_point" | 'CREATE TAG `only_point` (\n `geo` geography(point) NULL\n) ttl_duration = 0, ttl_col = ""' |
+
+  Scenario: insert geo data and query them
+    # Any geo shape(point/linestring/polygon) is allowed to insert to the column geography
+    When executing query:
+      """
+      INSERT VERTEX any_shape(geo) VALUES "101":(ST_GeogFromText("POINT(3 8)"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT VERTEX any_shape(geo) VALUES "102":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT VERTEX any_shape(geo) VALUES "103":(ST_GeogFromText("POLYGON((0 1, 1 2, 2 3, 0 1))"));
+      """
+    Then the execution should be successful
+    # Only point is allowed to insert to the column geograph(point)
+    When executing query:
+      """
+      INSERT VERTEX only_point(geo) VALUES "201":(ST_GeogFromText("POINT(3 8)"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT VERTEX only_point(geo) VALUES "202":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
+      """
+    Then a ExecutionError should be raised at runtime: Storage Error: The data type does not meet the requirements. Use the correct type of data.
+    When executing query:
+      """
+      INSERT VERTEX only_point(geo) VALUES "203":(ST_GeogFromText("POLYGON((0 1, 1 2, 2 3, 0 1))"));
+      """
+    Then a ExecutionError should be raised at runtime: Storage Error: The data type does not meet the requirements. Use the correct type of data.
+    # Only linestring is allowed to insert to the column geograph(linestring)
+    When executing query:
+      """
+      INSERT VERTEX only_linestring(geo) VALUES "301":(ST_GeogFromText("POINT(3 8)"));
+      """
+    Then a ExecutionError should be raised at runtime: Storage Error: The data type does not meet the requirements. Use the correct type of data.
+    When executing query:
+      """
+      INSERT VERTEX only_linestring(geo) VALUES "302":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT VERTEX only_linestring(geo) VALUES "303":(ST_GeogFromText("POLYGON((0 1, 1 2, 2 3, 0 1))"));
+      """
+    Then a ExecutionError should be raised at runtime: Storage Error: The data type does not meet the requirements. Use the correct type of data.
+    # Only polygon is allowed to insert to the column geograph(polygon)
+    When executing query:
+      """
+      INSERT VERTEX only_polygon(geo) VALUES "401":(ST_GeogFromText("POINT(3 8)"));
+      """
+    Then a ExecutionError should be raised at runtime: Storage Error: The data type does not meet the requirements. Use the correct type of data.
+    When executing query:
+      """
+      INSERT VERTEX only_polygon(geo) VALUES "402":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
+      """
+    Then a ExecutionError should be raised at runtime: Storage Error: The data type does not meet the requirements. Use the correct type of data.
+    When executing query:
+      """
+      INSERT VERTEX only_polygon(geo) VALUES "403":(ST_GeogFromText("POLYGON((0 1, 1 2, 2 3, 0 1))"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT EDGE any_shape_edge(geo) VALUES "201"->"302":(ST_GeogFromText("POLYGON((0 1, 1 2, 2 3, 0 1))"));
+      """
+    Then the execution should be successful
+    # Fetch the geo column
+    When executing query:
+      """
+      FETCH PROP ON any_shape "101","102","103" YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "101"    | "POINT(3 8)"                 |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      FETCH PROP ON only_point "201","202","203" YIELD ST_ASText(only_point.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_point.geo) |
+      | "201"    | "POINT(3 8)"              |
+    When executing query:
+      """
+      FETCH PROP ON only_linestring "301","302","303" YIELD ST_ASText(only_linestring.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_linestring.geo) |
+      | "302"    | "LINESTRING(3 8,4.7 73.23)"    |
+    When executing query:
+      """
+      FETCH PROP ON only_polygon "401","402","403" YIELD ST_ASText(only_polygon.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_polygon.geo)  |
+      | "403"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      FETCH PROP ON any_shape_edge "201"->"302" YIELD ST_ASText(any_shape_edge.geo);
+      """
+    Then the result should be, in any order:
+      | any_shape_edge._src | any_shape_edge._dst | any_shape_edge._rank | ST_ASText(any_shape_edge.geo) |
+      | "201"               | "302"               | 0                    | "POLYGON((0 1,1 2,2 3,0 1))"  |
+    # Create index on geo column
+    When executing query:
+      """
+      CREATE TAG INDEX any_shape_geo_index ON any_shape(geo);
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      CREATE TAG INDEX only_point_geo_index ON only_point(geo);
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      CREATE TAG INDEX only_linestring_geo_index ON only_linestring(geo);
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      CREATE TAG INDEX only_polygon_geo_index ON only_polygon(geo);
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      CREATE EDGE INDEX any_shape_edge_geo_index ON any_shape_edge(geo);
+      """
+    Then the execution should be successful
+    # Rebuild the geo index
+    When submit a job:
+      """
+      REBUILD TAG INDEX any_shape_geo_index;
+      """
+    Then wait the job to finish
+    When submit a job:
+      """
+      REBUILD TAG INDEX only_point_geo_index;
+      """
+    Then wait the job to finish
+    When submit a job:
+      """
+      REBUILD TAG INDEX only_linestring_geo_index;
+      """
+    Then wait the job to finish
+    When submit a job:
+      """
+      REBUILD TAG INDEX only_polygon_geo_index;
+      """
+    Then wait the job to finish
+    When submit a job:
+      """
+      REBUILD EDGE INDEX any_shape_edge_geo_index;
+      """
+    Then wait the job to finish
+    # Lookup on geo index
+    When executing query:
+      """
+      LOOKUP ON any_shape YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      LOOKUP ON only_point YIELD ST_ASText(only_point.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_point.geo) |
+      | "201"    | "POINT(3 8)"              |
+    When executing query:
+      """
+      LOOKUP ON only_linestring YIELD ST_ASText(only_linestring.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_linestring.geo) |
+      | "302"    | "LINESTRING(3 8,4.7 73.23)"    |
+    When executing query:
+      """
+      LOOKUP ON only_polygon YIELD ST_ASText(only_polygon.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_polygon.geo)  |
+      | "403"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      LOOKUP ON any_shape_edge YIELD ST_ASText(any_shape_edge.geo);
+      """
+    Then the result should be, in any order:
+      | SrcVID | DstVID | Ranking | ST_ASText(any_shape_edge.geo) |
+      | "201"  | "302"  | 0       | "POLYGON((0 1,1 2,2 3,0 1))"  |
+    # Insert geo data with index
+    When executing query:
+      """
+      INSERT VERTEX any_shape(geo) VALUES "108":(ST_GeogFromText("POINT(72.3 84.6)"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT VERTEX only_point(geo) VALUES "208":(ST_GeogFromText("POINT(0.01 0.01)"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT VERTEX only_linestring(geo) VALUES "308":(ST_GeogFromText("LINESTRING(9 9, 8 8, 7 7, 9 9)"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT VERTEX only_polygon(geo) VALUES "408":(ST_GeogFromText("POLYGON((0 1, 1 2, 2 3, 0 1))"));
+      """
+    Then the execution should be successful
+    When executing query:
+      """
+      INSERT EDGE any_shape_edge(geo) VALUES "108"->"408":(ST_GeogFromText("POLYGON((-20 -20, -20 20, 20 20, 20 -20, -20 -20), (1.0 1.0, 2.0 2.0, 0.0 2.0, 1.0 1.0))"));
+      """
+    Then the execution should be successful
+    # Lookup on geo index agagin
+    When executing query:
+      """
+      LOOKUP ON any_shape YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+      | "108"    | "POINT(72.3 84.6)"           |
+    When executing query:
+      """
+      LOOKUP ON only_point YIELD ST_ASText(only_point.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_point.geo) |
+      | "201"    | "POINT(3 8)"              |
+      | "208"    | "POINT(0.01 0.01)"        |
+    When executing query:
+      """
+      LOOKUP ON only_linestring YIELD ST_ASText(only_linestring.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_linestring.geo) |
+      | "302"    | "LINESTRING(3 8,4.7 73.23)"    |
+      | "308"    | "LINESTRING(9 9,8 8,7 7,9 9)"  |
+    When executing query:
+      """
+      LOOKUP ON only_polygon YIELD ST_ASText(only_polygon.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(only_polygon.geo)  |
+      | "403"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+      | "408"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      LOOKUP ON any_shape_edge YIELD ST_ASText(any_shape_edge.geo);
+      """
+    Then the result should be, in any order:
+      | SrcVID | DstVID | Ranking | ST_ASText(any_shape_edge.geo)                                      |
+      | "108"  | "408"  | 0       | "POLYGON((-20 -20,-20 20,20 20,20 -20,-20 -20),(1 1,2 2,0 2,1 1))" |
+      | "201"  | "302"  | 0       | "POLYGON((0 1,1 2,2 3,0 1))"                                       |
+    # Lookup and Yield geo functions
+    When executing query:
+      """
+      LOOKUP ON any_shape YIELD S2_CellIdFromPoint(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | S2_CellIdFromPoint(any_shape.geo) |
+      | "101"    | 1166542697063163289               |
+      | "102"    | BAD_DATA                          |
+      | "103"    | BAD_DATA                          |
+      | "108"    | 4987215245349669805               |
+    When executing query:
+      """
+      LOOKUP ON any_shape YIELD S2_CoveringCellIds(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | S2_CoveringCellIds(any_shape.geo)                                                                                                                                        |
+      | "101"    | [1166542697063163289]                                                                                                                                                    |
+      | "102"    | [1167558203395801088, 1279022294173220864, 1315051091192184832, 1351079888211148800, 5039527983027585024, 5062045981164437504, 5174635971848699904, 5183643171103440896] |
+      | "103"    | [1152391494368201343, 1153466862374223872, 1153554823304445952, 1153836298281156608, 1153959443583467520, 1154240918560178176, 1160503736791990272, 1160591697722212352] |
+      | "108"    | [4987215245349669805]                                                                                                                                                    |
+    # Lookup with geo predicates which could be index accelerated
+    # ST_Intersects
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Intersects(any_shape.geo, ST_GeogFromText('POINT(3 8)')) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)    |
+      | "101"    | "POINT(3 8)"                |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Intersects(any_shape.geo, ST_GeogFromText('POINT(0 1)')) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Intersects(any_shape.geo, ST_GeogFromText('POINT(4.7 73.23)')) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)    |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Intersects(any_shape.geo, ST_Point(72.3, 84.6)) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo) |
+      | "108"    | "POINT(72.3 84.6)"       |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Intersects(ST_Point(72.3, 84.6), any_shape.geo) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo) |
+      | "108"    | "POINT(72.3 84.6)"       |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Intersects(any_shape.geo, any_shape.geo) YIELD ST_ASText(any_shape.geo);
+      """
+    Then a SemanticError should be raised at runtime: Expression ST_Intersects(any_shape.geo,any_shape.geo) not supported yet
+    # ST_Distance
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Distance(any_shape.geo, ST_Point(3, 8)) < 1.0 YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)    |
+      | "101"    | "POINT(3 8)"                |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Distance(any_shape.geo, ST_Point(3, 8)) <= 1.0 YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)    |
+      | "101"    | "POINT(3 8)"                |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Distance(any_shape.geo, ST_Point(3, 8)) <= 8909524.383934561 YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+      | "108"    | "POINT(72.3 84.6)"           |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Distance(any_shape.geo, ST_Point(3, 8)) < 8909524.383934561 YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Distance(any_shape.geo, ST_Point(3, 8)) < 8909524.383934563 YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+      | "108"    | "POINT(72.3 84.6)"           |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE 8909524.383934560 > ST_Distance(any_shape.geo, ST_Point(3, 8)) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE 8909524.3839345630 >= ST_Distance(any_shape.geo, ST_Point(3, 8)) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+      | "108"    | "POINT(72.3 84.6)"           |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Distance(any_shape.geo, ST_Point(3, 8)) > 1.0 YIELD ST_ASText(any_shape.geo);
+      """
+    Then a SemanticError should be raised at runtime: Expression (ST_Distance(any_shape.geo,ST_Point(3,8))>1) not supported yet
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Distance(any_shape.geo, ST_Point(3, 8)) != 1.0 YIELD ST_ASText(any_shape.geo);
+      """
+    Then a SemanticError should be raised at runtime: Expression (ST_Distance(any_shape.geo,ST_Point(3,8))!=1) not supported yet
+    # ST_DWithin
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_DWithin(any_shape.geo, ST_Point(3, 8), 8909524.383934561) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "101"    | "POINT(3 8)"                 |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)"  |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+      | "108"    | "POINT(72.3 84.6)"           |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_DWithin(any_shape.geo, ST_Point(3, 8), 100.0) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)    |
+      | "101"    | "POINT(3 8)"                |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)" |
+    # ST_Covers
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Covers(any_shape.geo, ST_Point(3, 8)) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)    |
+      | "101"    | "POINT(3 8)"                |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Covers(any_shape.geo, ST_Point(3, 8)) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)    |
+      | "101"    | "POINT(3 8)"                |
+      | "102"    | "LINESTRING(3 8,4.7 73.23)" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_Covers(ST_GeogFromText('POLYGON((-0.7 3.8,3.6 3.2,1.8 -0.8,-3.4 2.4,-0.7 3.8))'), any_shape.geo) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
+    When executing query:
+      """
+      LOOKUP ON any_shape WHERE ST_CoveredBy(any_shape.geo, ST_GeogFromText('POLYGON((-0.7 3.8,3.6 3.2,1.8 -0.8,-3.4 2.4,-0.7 3.8))')) YIELD ST_ASText(any_shape.geo);
+      """
+    Then the result should be, in any order:
+      | VertexID | ST_ASText(any_shape.geo)     |
+      | "103"    | "POLYGON((0 1,1 2,2 3,0 1))" |
diff --git a/tests/tck/features/geo/GeoIndex.feature b/tests/tck/features/geo/GeoIndex.feature
new file mode 100644
index 00000000000..e69de29bb2d