code: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   *
   */
  
  package org.apache.skywalking.oap.server.storage.plugin.banyandb;
  
  import com.google.gson.JsonObject;
  import lombok.RequiredArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
  import org.apache.skywalking.banyandb.v1.client.DataPoint;
  import org.apache.skywalking.banyandb.v1.client.MeasureWrite;
  import org.apache.skywalking.banyandb.v1.client.RowEntity;
  import org.apache.skywalking.banyandb.v1.client.StreamWrite;
  import org.apache.skywalking.banyandb.v1.client.TagAndValue;
  import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
  import org.apache.skywalking.banyandb.v1.client.metadata.Serializable;
  import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
  import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
  import org.apache.skywalking.oap.server.core.analysis.record.Record;
  import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity;
  import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage;
  import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject;
  import org.apache.skywalking.oap.server.storage.plugin.banyandb.util.ByteUtil;
  
  import java.util.List;
  
  public class BanyanDBConverter {
  
      public static final String ID = "id";
  
      public static class StorageToStream implements Convert2Entity {
          private final MetadataRegistry.Schema schema;
          private final RowEntity rowEntity;
  
          public StorageToStream(String streamModelName, RowEntity rowEntity) {
              this.schema = MetadataRegistry.INSTANCE.findRecordMetadata(streamModelName);
              this.rowEntity = rowEntity;
          }
  
          @Override
          public Object get(String fieldName) {
              if (fieldName.equals(Record.TIME_BUCKET)) {
                  final String timestampColumnName = schema.getTimestampColumn4Stream();
                  long timestampMillis = ((Number) this.get(timestampColumnName)).longValue();
                  return TimeBucket.getTimeBucket(timestampMillis, schema.getMetadata().getDownSampling());
              }
              MetadataRegistry.ColumnSpec spec = schema.getSpec(fieldName);
              if (double.class.equals(spec.getColumnClass())) {
                  return ByteUtil.bytes2Double(rowEntity.getTagValue(fieldName));
              } else {
                  return rowEntity.getTagValue(fieldName);
              }
          }
  
          @Override
          public byte[] getBytes(String fieldName) {
              return rowEntity.getTagValue(fieldName);
          }
      }
  
      @Slf4j
      @RequiredArgsConstructor
      public static class StreamToStorage implements Convert2Storage<StreamWrite> {
          private final MetadataRegistry.Schema schema;
          private final StreamWrite streamWrite;
  
          @Override
          public void accept(String fieldName, Object fieldValue) {
              if (fieldName.equals(Record.TIME_BUCKET)) {
                  return;
              }
              if (fieldName.equals(this.schema.getTimestampColumn4Stream())) {
                  streamWrite.setTimestamp((long) fieldValue);
              }
              MetadataRegistry.ColumnSpec columnSpec = this.schema.getSpec(fieldName);
              if (columnSpec == null) {
                  throw new IllegalArgumentException("fail to find tag[" + fieldName + "]");
              }
              if (columnSpec.getColumnType() != MetadataRegistry.ColumnType.TAG) {
                  throw new IllegalArgumentException("ColumnType other than TAG is not supported for Stream, " +
                          "it should be an internal error, please submit an issue to the SkyWalking community");
              }
              try {
                  this.streamWrite.tag(fieldName, buildTag(fieldValue, columnSpec.getColumnClass()));
              } catch (BanyanDBException ex) {
                  log.error("fail to add tag", ex);
              }
          }
  
          @Override
          public void accept(String fieldName, byte[] fieldValue) {
              try {
                  this.streamWrite.tag(fieldName, TagAndValue.binaryTagValue(fieldValue));
              } catch (BanyanDBException ex) {
                  log.error("fail to add tag", ex);
              }
          }
  
          @Override
          public void accept(String fieldName, List<String> fieldValue) {
              try {
                  this.streamWrite.tag(fieldName, TagAndValue.stringArrayTagValue(fieldValue));
              } catch (BanyanDBException ex) {
                  log.error("fail to accept string array tag", ex);
              }
          }
  
          @Override
          public Object get(String fieldName) {
              throw new IllegalStateException("should not reach here");
          }
  
          @Override
          public StreamWrite obtain() {
              return this.streamWrite;
          }
      }
  
      @Slf4j
      @RequiredArgsConstructor
      public static class MeasureToStorage implements Convert2Storage<MeasureWrite> {
          private final MetadataRegistry.Schema schema;
          private final MeasureWrite measureWrite;
  
          @Override
          public void accept(String fieldName, Object fieldValue) {
              if (fieldName.equals(Metrics.TIME_BUCKET)) {
                  return;
              }
              MetadataRegistry.ColumnSpec columnSpec = this.schema.getSpec(fieldName);
              if (columnSpec == null) {
                  throw new IllegalArgumentException("fail to find tag/field[" + fieldName + "]");
              }
              try {
                  if (columnSpec.getColumnType() == MetadataRegistry.ColumnType.TAG) {
                      this.measureWrite.tag(fieldName, buildTag(fieldValue, columnSpec.getColumnClass()));
                  } else {
                      this.measureWrite.field(fieldName, buildField(fieldValue, columnSpec.getColumnClass()));
                  }
              } catch (BanyanDBException ex) {
                  log.error("fail to add tag/field", ex);
              }
          }
  
          public void acceptID(String id) {
              try {
                  this.measureWrite.tag(ID, TagAndValue.stringTagValue(id));
              } catch (BanyanDBException ex) {
                  log.error("fail to add ID tag", ex);
              }
          }
  
          @Override
          public void accept(String fieldName, byte[] fieldValue) {
              MetadataRegistry.ColumnSpec columnSpec = this.schema.getSpec(fieldName);
              try {
                  if (columnSpec.getColumnType() == MetadataRegistry.ColumnType.TAG) {
                      this.measureWrite.tag(fieldName, TagAndValue.binaryTagValue(fieldValue));
                  } else {
                      this.measureWrite.field(fieldName, TagAndValue.binaryFieldValue(fieldValue));
                  }
              } catch (BanyanDBException ex) {
                  log.error("fail to add binary tag/field", ex);
              }
          }
  
          @Override
          public void accept(String fieldName, List<String> fieldValue) {
              try {
                  this.measureWrite.tag(fieldName, TagAndValue.stringArrayTagValue(fieldValue));
              } catch (BanyanDBException ex) {
                  log.error("fail to accept string array tag", ex);
              }
          }
  
          @Override
          public Object get(String fieldName) {
              throw new IllegalStateException("should not reach here");
          }
  
          @Override
          public MeasureWrite obtain() {
              return this.measureWrite;
          }
      }
  
      private static Serializable<BanyandbModel.TagValue> buildTag(Object value, final Class<?> clazz) {
          if (int.class.equals(clazz) || Integer.class.equals(clazz)) {
              return TagAndValue.longTagValue(((Number) value).longValue());
          } else if (Long.class.equals(clazz) || long.class.equals(clazz)) {
              return TagAndValue.longTagValue((Long) value);
          } else if (String.class.equals(clazz)) {
              return TagAndValue.stringTagValue((String) value);
          } else if (Double.class.equals(clazz) || double.class.equals(clazz)) {
              return TagAndValue.binaryTagValue(ByteUtil.double2Bytes((double) value));
          } else if (StorageDataComplexObject.class.isAssignableFrom(clazz)) {
              return TagAndValue.stringTagValue(((StorageDataComplexObject<?>) value).toStorageData());
          } else if (clazz.isEnum()) {
              return TagAndValue.longTagValue((int) value);
          } else if (JsonObject.class.equals(clazz)) {
              return TagAndValue.stringTagValue((String) value);
          } else if (byte[].class.equals(clazz)) {
              return TagAndValue.stringTagValue((String) value);
          }
          throw new IllegalStateException(clazz.getSimpleName() + " is not supported");
      }
  
      private static Serializable<BanyandbModel.FieldValue> buildField(Object value, final Class<?> clazz) {
          if (Integer.class.equals(clazz) || int.class.equals(clazz)) {
              return TagAndValue.longFieldValue(((Number) value).longValue());
          } else if (Long.class.equals(clazz) || long.class.equals(clazz)) {
              return TagAndValue.longFieldValue((Long) value);
          } else if (String.class.equals(clazz)) {
              return TagAndValue.stringFieldValue((String) value);
          } else if (Double.class.equals(clazz) || double.class.equals(clazz)) {
              return TagAndValue.binaryFieldValue(ByteUtil.double2Bytes((double) value));
          } else if (StorageDataComplexObject.class.isAssignableFrom(clazz)) {
              return TagAndValue.stringFieldValue(((StorageDataComplexObject<?>) value).toStorageData());
          }
          throw new IllegalStateException(clazz.getSimpleName() + " is not supported");
      }
  
      public static class StorageToMeasure implements Convert2Entity {
          private final MetadataRegistry.Schema schema;
          private final DataPoint dataPoint;
  
          public StorageToMeasure(MetadataRegistry.Schema schema, DataPoint dataPoint) {
              this.schema = schema;
              this.dataPoint = dataPoint;
          }
  
          @Override
          public Object get(String fieldName) {
              if (fieldName.equals(Metrics.TIME_BUCKET)) {
                  return TimeBucket.getTimeBucket(dataPoint.getTimestamp(), schema.getMetadata().getDownSampling());
              }
              MetadataRegistry.ColumnSpec spec = schema.getSpec(fieldName);
              switch (spec.getColumnType()) {
                  case TAG:
                      if (double.class.equals(spec.getColumnClass())) {
                          return ByteUtil.bytes2Double(dataPoint.getTagValue(fieldName));
                      } else {
                          return dataPoint.getTagValue(fieldName);
                      }
                  case FIELD:
                  default:
                      if (double.class.equals(spec.getColumnClass())) {
                          return ByteUtil.bytes2Double(dataPoint.getFieldValue(fieldName));
                      } else {
                          return dataPoint.getFieldValue(fieldName);
                      }
              }
          }
  
          @Override
          public byte[] getBytes(String fieldName) {
              // TODO: double may be a field?
              return dataPoint.getFieldValue(fieldName);
          }
      }
  }
  