/**
 * Copyright 2020 Alibaba Group Holding Limited.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef ANALYTICAL_ENGINE_HTAP_HTAP_DS_IMPL_H_
#define ANALYTICAL_ENGINE_HTAP_HTAP_DS_IMPL_H_

#include <utility>

#include "vineyard/client/client.h"
#include "vineyard/graph/fragment/arrow_fragment.h"

#include "global_store_ffi.h"
#include "graph_schema.h"

namespace vineyard {

namespace htap_impl {

using VID_TYPE = uint64_t;
using OID_TYPE = int64_t;
using INT32_OID_TYPE = int32_t;
using STRING_OID_TYPE = std::string;

using FRAGMENT_TYPE = ::vineyard::ArrowFragment<OID_TYPE, VID_TYPE>;
using INT32_FRAGMENT_TYPE = ::vineyard::ArrowFragment<INT32_OID_TYPE, VID_TYPE>;
using STRING_FRAGMENT_TYPE = ::vineyard::ArrowFragment<STRING_OID_TYPE, VID_TYPE>;

using VERTEX_MAP_TYPE = ::vineyard::ArrowVertexMap<FRAGMENT_TYPE::internal_oid_t, VID_TYPE>;
using INT32_VERTEX_MAP_TYPE = ::vineyard::ArrowVertexMap<INT32_FRAGMENT_TYPE::internal_oid_t, VID_TYPE>;
using STRING_VERTEX_MAP_TYPE = ::vineyard::ArrowVertexMap<STRING_FRAGMENT_TYPE::internal_oid_t, VID_TYPE>;

using FRAG_ID_TYPE = ::vineyard::fid_t;
using EID_TYPE = typename FRAGMENT_TYPE::eid_t;
using VERTEX_RANGE_TYPE = std::pair<VID_TYPE, VID_TYPE>;
using VERTEX_TYPE = typename FRAGMENT_TYPE::vertex_t;

struct GraphHandleImpl {
  vineyard::Client* client = nullptr;

  bool use_int64_oid = true;
  bool use_int32_oid = true;
  bool use_string_oid = false;

  FRAGMENT_TYPE* fragments = nullptr;
  INT32_FRAGMENT_TYPE* int32_fragments = nullptr;
  STRING_FRAGMENT_TYPE* string_fragments = nullptr;

  VERTEX_MAP_TYPE* vertex_map = nullptr;
  INT32_VERTEX_MAP_TYPE* int32_vertex_map = nullptr;
  STRING_VERTEX_MAP_TYPE* string_vertex_map = nullptr;

  FRAG_ID_TYPE fnum;
  vineyard::IdParser<VID_TYPE> vid_parser;
  vineyard::IdParser<EID_TYPE> eid_parser;
  vineyard::htap::MGPropertyGraphSchema* schema = nullptr;
  FRAG_ID_TYPE local_fnum;
  FRAG_ID_TYPE* local_fragments = nullptr;
  int vertex_label_num;
  int edge_label_num;

  PartitionId channel_num;
  VID_TYPE** vertex_chunk_sizes = nullptr;
};

inline int get_edge_partition_id(EID_TYPE id, GraphHandleImpl* handle) {
  return handle->eid_parser.GetFid(id);
}

void get_graph_handle(ObjectId id, PartitionId channel_num,
                      GraphHandleImpl* handle);

void free_graph_handle(GraphHandleImpl* handle);

struct GetVertexIteratorImpl {
  VID_TYPE* ids;
  int count;
  int index;
};

template <typename FRAGMENT_TYPE>
void get_vertices(FRAGMENT_TYPE* frag, LabelId* label, VertexId* ids, int count,
                  GetVertexIteratorImpl* out);

void free_get_vertex_iterator(GetVertexIteratorImpl* iter);

int get_vertices_next(GetVertexIteratorImpl* iter, Vertex* v_out);

struct GetAllVerticesIteratorImpl {
  VERTEX_RANGE_TYPE* ranges;
  int range_num;
  int range_id;

  VID_TYPE cur_vertex_id;
};

template <typename FRAGMENT_TYPE>
void get_all_vertices(FRAGMENT_TYPE* frag, PartitionId channel_id,
                      const VID_TYPE* chunk_sizes, LabelId* labels,
                      int labels_count, int64_t limit,
                      GetAllVerticesIteratorImpl* out);

void free_get_all_vertices_iterator(GetAllVerticesIteratorImpl* iter);

int get_all_vertices_next(GetAllVerticesIteratorImpl* iter, Vertex* v_out);

struct PropertiesIteratorImpl {
  GraphHandleImpl* handle;
  arrow::Table* table;
  bool vertex_or_edge;  // true: vertex, false: edge
  LabelId label_id;
  int64_t row_id;
  PropertyId col_num;
  PropertyId col_id;
};

template <typename FRAGMENT_TYPE>
typename FRAGMENT_TYPE::oid_t get_outer_id(FRAGMENT_TYPE* frag, Vertex v);

template <typename FRAGMENT_TYPE>
int get_vertex_property(FRAGMENT_TYPE* frag, Vertex v, PropertyId id,
                        Property* p_out);

template <typename FRAGMENT_TYPE>
void get_vertex_properties(FRAGMENT_TYPE* frag, Vertex v,
                           PropertiesIteratorImpl* iter);

using NBR_TYPE = typename FRAGMENT_TYPE::nbr_unit_t;
// using ADJ_LIST_TYPE = std::pair<const NBR_TYPE*, const NBR_TYPE*>;
struct AdjListUnit {
  const NBR_TYPE* begin;
  const NBR_TYPE* end;
  LabelId label;
} __attribute__((packed));

struct EdgeIteratorImpl {
  // FRAG_ID_TYPE fid;
  FRAGMENT_TYPE* fragment = nullptr;
  INT32_FRAGMENT_TYPE* int32_fragment = nullptr;
  STRING_FRAGMENT_TYPE* string_fragment = nullptr;
  vineyard::IdParser<EID_TYPE>* eid_parser;

  int64_t src;
  AdjListUnit* lists;
  int list_num;

  int list_id;
  const NBR_TYPE* cur_edge;
};

void empty_edge_iterator(EdgeIteratorImpl* iter);

template <typename FRAGMENT_TYPE>
void get_out_edges(FRAGMENT_TYPE* frag,
                   vineyard::IdParser<EID_TYPE>* eid_parser, VertexId src_id,
                   LabelId* labels, int labels_count, int64_t limit,
                   EdgeIteratorImpl* iter);

int out_edge_next(EdgeIteratorImpl* iter, Edge* e_out);

template <typename FRAGMENT_TYPE>
void get_in_edges(FRAGMENT_TYPE* frag, vineyard::IdParser<EID_TYPE>* eid_parser,
                  VertexId dst_id, LabelId* labels, int labels_count,
                  int64_t limit, EdgeIteratorImpl* iter);

int in_edge_next(EdgeIteratorImpl* iter, Edge* e_out);

struct GetAllEdgesIteratorImpl {
  FRAGMENT_TYPE* fragment = nullptr;
  INT32_FRAGMENT_TYPE* int32_fragment = nullptr;
  STRING_FRAGMENT_TYPE* string_fragment = nullptr;
  LabelId* e_labels;
  vineyard::IdParser<EID_TYPE>* eid_parser;
  int e_labels_count;

  int cur_v_label;
  VERTEX_RANGE_TYPE cur_range;

  EdgeIteratorImpl ei;

  const VID_TYPE* chunk_sizes;
  PartitionId channel_id;

  int64_t index;
  int64_t limit;
};

template <typename FRAGMENT_TYPE>
void get_all_edges(FRAGMENT_TYPE* frag, PartitionId channel_id,
                   const VID_TYPE* chunk_sizes,
                   vineyard::IdParser<EID_TYPE>* eid_parser, LabelId* labels,
                   int labels_count, int64_t limit,
                   GetAllEdgesIteratorImpl* iter);

int get_all_edges_next(GetAllEdgesIteratorImpl* iter, Edge* e_out);

void free_edge_iterator(EdgeIteratorImpl* iter);

void free_get_all_edges_iterator(GetAllEdgesIteratorImpl* iter);

template <typename FRAGMENT_TYPE>
EdgeId get_edge_id(FRAGMENT_TYPE* frag, LabelId label, int64_t offset);

template <typename FRAGMENT_TYPE>
int get_edge_property(FRAGMENT_TYPE* frag, LabelId label, int64_t offset,
                      PropertyId id, Property* p_out);

template <typename FRAGMENT_TYPE>
void get_edge_properties(FRAGMENT_TYPE* frag, LabelId label, int64_t offset,
                         PropertiesIteratorImpl* iter);

int properties_next(PropertiesIteratorImpl* iter, Property* p_out);

void free_properties_iterator(PropertiesIteratorImpl* iter);

union PodProperties {
  bool bool_value;
  char char_value;
  int16_t int16_value;
  int int_value;
  int64_t long_value;
  float float_value;
  double double_value;
};

int get_property_as_bool(Property* property, bool* out);
int get_property_as_char(Property* property, char* out);
int get_property_as_short(Property* property, int16_t* out);
int get_property_as_int(Property* property, int* out);
int get_property_as_long(Property* property, int64_t* out);
int get_property_as_float(Property* property, float* out);
int get_property_as_double(Property* property, double* out);
int get_property_as_string(Property* property, const char** out, int* out_len);
int get_property_as_bytes(Property* property, const char** out, int* out_len);
int get_property_as_int_list(Property* property, const int** out, int* out_len);
int get_property_as_long_list(Property* property, const int64_t** out,
                              int* out_len);
int get_property_as_float_list(Property* property, const float** out,
                               int* out_len);
int get_property_as_double_list(Property* property, const double** out,
                                int* out_len);
int get_property_as_string_list(Property* property, const char*** out,
                                const int** out_len, int* out_num);
int get_property_as_date32(struct Property* property, int32_t *out);
int get_property_as_date64(struct Property* property, int64_t *out);
int get_property_as_time32_s(struct Property* property, int32_t *out);
int get_property_as_time32_ms(struct Property* property, int32_t *out);
int get_property_as_time32_us(struct Property* property, int32_t *out);
int get_property_as_time32_ns(struct Property* property, int32_t *out);
int get_property_as_time64_s(struct Property* property, int64_t *out);
int get_property_as_time64_ms(struct Property* property, int64_t *out);
int get_property_as_time64_us(struct Property* property, int64_t *out);
int get_property_as_time64_ns(struct Property* property, int64_t *out);
int get_property_as_timestamp_s(struct Property* property, int64_t *out,
                                const char **out_timezone, int *out_timezone_len);
int get_property_as_timestamp_ms(struct Property* property, int64_t *out,
                                 const char **out_timezone, int *out_timezone_len);
int get_property_as_timestamp_us(struct Property* property, int64_t *out,
                                 const char **out_timezone, int *out_timezone_len);
int get_property_as_timestamp_ns(struct Property* property, int64_t *out,
                                 const char **out_timezone, int *out_timezone_len);

void free_property(Property* property);

}  // namespace htap_impl
}  // namespace vineyard

#endif  // ANALYTICAL_ENGINE_HTAP_HTAP_DS_IMPL_H_
