content: /*
   * 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.asterix.metadata.entities;
  
  import java.io.DataOutput;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Objects;
  
  import org.apache.asterix.builders.IARecordBuilder;
  import org.apache.asterix.builders.OrderedListBuilder;
  import org.apache.asterix.builders.RecordBuilder;
  import org.apache.asterix.common.config.DatasetConfig.DatasetType;
  import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
  import org.apache.asterix.metadata.IDatasetDetails;
  import org.apache.asterix.metadata.bootstrap.DatasetEntity;
  import org.apache.asterix.metadata.bootstrap.MetadataRecordTypes;
  import org.apache.asterix.om.base.ABoolean;
  import org.apache.asterix.om.base.AInt8;
  import org.apache.asterix.om.base.AMutableInt8;
  import org.apache.asterix.om.base.AMutableString;
  import org.apache.asterix.om.base.AString;
  import org.apache.asterix.om.types.AOrderedListType;
  import org.apache.asterix.om.types.ARecordType;
  import org.apache.asterix.om.types.BuiltinType;
  import org.apache.asterix.om.types.IAType;
  import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
  
  public class InternalDatasetDetails implements IDatasetDetails {
  
      private static final long serialVersionUID = 1L;
  
      public enum FileStructure {
          BTREE
      }
  
      public enum PartitioningStrategy {
          HASH
      }
  
      private final FileStructure fileStructure;
      private final PartitioningStrategy partitioningStrategy;
      private final List<List<String>> partitioningKeys;
      private final List<List<String>> primaryKeys;
      private final List<IAType> primaryKeyTypes;
      private final boolean autogenerated;
      private final Integer filterSourceIndicator;
      private final List<String> filterField;
      private final List<Integer> keySourceIndicators;
      private final boolean isDatasetWithoutTypeSpecification;
  
      public static final String FILTER_FIELD_NAME = "FilterField";
      public static final String FILTER_SOURCE_INDICATOR_FIELD_NAME = "FilterSourceIndicator";
      public static final String KEY_FILD_SOURCE_INDICATOR_FIELD_NAME = "KeySourceIndicator";
      public static final String PRIMARY_KEY_TYPES_FIELD_NAME = "PrimaryKeyTypes";
  
      public InternalDatasetDetails(FileStructure fileStructure, PartitioningStrategy partitioningStrategy,
              List<List<String>> partitioningKey, List<List<String>> primaryKey, List<Integer> keyFieldIndicators,
              List<IAType> primaryKeyType, boolean autogenerated, Integer filterSourceIndicator, List<String> filterField,
              boolean isDatasetWithoutTypeSpecification) {
          this.fileStructure = fileStructure;
          this.partitioningStrategy = partitioningStrategy;
          this.partitioningKeys = partitioningKey;
          this.primaryKeys = primaryKey;
          if (keyFieldIndicators == null) {
              // Create a dummy list.
              keyFieldIndicators = new ArrayList<>();
              for (int index = 0; index < partitioningKey.size(); ++index) {
                  keyFieldIndicators.add(0);
              }
          }
          this.keySourceIndicators = keyFieldIndicators;
          this.primaryKeyTypes = primaryKeyType;
          this.autogenerated = autogenerated;
          if (filterSourceIndicator != null) {
              // to make sure filter source indicator and filter field is consistent
              this.filterSourceIndicator = filterSourceIndicator;
              this.filterField = Objects.requireNonNull(filterField);
          } else {
              this.filterSourceIndicator = null;
              this.filterField = null;
          }
          this.isDatasetWithoutTypeSpecification = isDatasetWithoutTypeSpecification;
      }
  
      public InternalDatasetDetails(FileStructure fileStructure, PartitioningStrategy partitioningStrategy,
              List<List<String>> partitioningKey, List<List<String>> primaryKey, List<Integer> keyFieldIndicators,
              List<IAType> primaryKeyType, boolean autogenerated, Integer filterSourceIndicator,
              List<String> filterField) {
          this(fileStructure, partitioningStrategy, partitioningKey, primaryKey, keyFieldIndicators, primaryKeyType,
                  autogenerated, filterSourceIndicator, filterField, false);
      }
  
      public List<List<String>> getPartitioningKey() {
          return partitioningKeys;
      }
  
      public boolean isAutogenerated() {
          return autogenerated;
      }
  
      public List<List<String>> getPrimaryKey() {
          return primaryKeys;
      }
  
      public List<Integer> getKeySourceIndicator() {
          return keySourceIndicators;
      }
  
      public List<IAType> getPrimaryKeyType() {
          return primaryKeyTypes;
      }
  
      public FileStructure getFileStructure() {
          return fileStructure;
      }
  
      public PartitioningStrategy getPartitioningStrategy() {
          return partitioningStrategy;
      }
  
      public List<String> getFilterField() {
          return filterField;
      }
  
      public Integer getFilterSourceIndicator() {
          return filterSourceIndicator;
      }
  
      public boolean isDatasetWithoutTypeSpecification() {
          return isDatasetWithoutTypeSpecification;
      }
  
      @Override
      public DatasetType getDatasetType() {
          return DatasetType.INTERNAL;
      }
  
      @Override
      public void writeDatasetDetailsRecordType(DataOutput out, DatasetEntity datasetEntity) throws HyracksDataException {
  
          IARecordBuilder internalRecordBuilder = new RecordBuilder();
          OrderedListBuilder listBuilder = new OrderedListBuilder();
          ArrayBackedValueStorage fieldValue = new ArrayBackedValueStorage();
          ArrayBackedValueStorage itemValue = new ArrayBackedValueStorage();
          OrderedListBuilder primaryKeyListBuilder = new OrderedListBuilder();
          AOrderedListType stringList = new AOrderedListType(BuiltinType.ASTRING, null);
          AOrderedListType int8List = new AOrderedListType(BuiltinType.AINT8, null);
          AOrderedListType heterogeneousList = new AOrderedListType(BuiltinType.ANY, null);
          internalRecordBuilder.reset(MetadataRecordTypes.INTERNAL_DETAILS_RECORDTYPE);
          AMutableString aString = new AMutableString("");
          AMutableInt8 aInt8 = new AMutableInt8((byte) 0);
          @SuppressWarnings("unchecked")
          ISerializerDeserializer<ABoolean> booleanSerde =
                  SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ABOOLEAN);
          @SuppressWarnings("unchecked")
          ISerializerDeserializer<AString> stringSerde =
                  SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ASTRING);
          @SuppressWarnings("unchecked")
          ISerializerDeserializer<AInt8> int8Serde =
                  SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT8);
  
          // write field 0
          fieldValue.reset();
          aString.setValue(getFileStructure().toString());
          stringSerde.serialize(aString, fieldValue.getDataOutput());
          internalRecordBuilder.addField(MetadataRecordTypes.INTERNAL_DETAILS_ARECORD_FILESTRUCTURE_FIELD_INDEX,
                  fieldValue);
  
          // write field 1
          fieldValue.reset();
          aString.setValue(getPartitioningStrategy().toString());
          stringSerde.serialize(aString, fieldValue.getDataOutput());
          internalRecordBuilder.addField(MetadataRecordTypes.INTERNAL_DETAILS_ARECORD_PARTITIONSTRATEGY_FIELD_INDEX,
                  fieldValue);
  
          // write field 2
          primaryKeyListBuilder.reset((AOrderedListType) MetadataRecordTypes.INTERNAL_DETAILS_RECORDTYPE
                  .getFieldTypes()[MetadataRecordTypes.INTERNAL_DETAILS_ARECORD_PARTITIONKEY_FIELD_INDEX]);
          for (List<String> field : partitioningKeys) {
              listBuilder.reset(stringList);
              for (String subField : field) {
                  itemValue.reset();
                  aString.setValue(subField);
                  stringSerde.serialize(aString, itemValue.getDataOutput());
                  listBuilder.addItem(itemValue);
              }
              itemValue.reset();
              listBuilder.write(itemValue.getDataOutput(), true);
              primaryKeyListBuilder.addItem(itemValue);
          }
          fieldValue.reset();
          primaryKeyListBuilder.write(fieldValue.getDataOutput(), true);
          internalRecordBuilder.addField(MetadataRecordTypes.INTERNAL_DETAILS_ARECORD_PARTITIONKEY_FIELD_INDEX,
                  fieldValue);
  
          // write field 3
          primaryKeyListBuilder.reset((AOrderedListType) MetadataRecordTypes.INTERNAL_DETAILS_RECORDTYPE
                  .getFieldTypes()[MetadataRecordTypes.INTERNAL_DETAILS_ARECORD_PRIMARYKEY_FIELD_INDEX]);
          for (List<String> field : primaryKeys) {
              listBuilder.reset(stringList);
              for (String subField : field) {
                  itemValue.reset();
                  aString.setValue(subField);
                  stringSerde.serialize(aString, itemValue.getDataOutput());
                  listBuilder.addItem(itemValue);
              }
              itemValue.reset();
              listBuilder.write(itemValue.getDataOutput(), true);
              primaryKeyListBuilder.addItem(itemValue);
          }
          fieldValue.reset();
          primaryKeyListBuilder.write(fieldValue.getDataOutput(), true);
          internalRecordBuilder.addField(MetadataRecordTypes.INTERNAL_DETAILS_ARECORD_PRIMARYKEY_FIELD_INDEX, fieldValue);
  
          // write field 4
          fieldValue.reset();
          ABoolean b = isAutogenerated() ? ABoolean.TRUE : ABoolean.FALSE;
          booleanSerde.serialize(b, fieldValue.getDataOutput());
          internalRecordBuilder.addField(MetadataRecordTypes.INTERNAL_DETAILS_ARECORD_AUTOGENERATED_FIELD_INDEX,
                  fieldValue);
  
          // Serialize Primary Key types if available and if dataset without type specification
          if (isDatasetWithoutTypeSpecification && primaryKeyTypes != null && !primaryKeyTypes.isEmpty()) {
              ArrayBackedValueStorage nameValue = new ArrayBackedValueStorage();
              nameValue.reset();
              aString.setValue(PRIMARY_KEY_TYPES_FIELD_NAME);
              stringSerde.serialize(aString, nameValue.getDataOutput());
  
              listBuilder.reset(AOrderedListType.FULL_OPEN_ORDEREDLIST_TYPE);
              for (IAType keyType : primaryKeyTypes) {
                  itemValue.reset();
                  aString.setValue(keyType.getTypeName());
                  stringSerde.serialize(aString, itemValue.getDataOutput());
                  listBuilder.addItem(itemValue);
              }
              fieldValue.reset();
              listBuilder.write(fieldValue.getDataOutput(), true);
              internalRecordBuilder.addField(nameValue, fieldValue);
          }
  
          // write filter fields if any
          Integer filterSourceIndicator = getFilterSourceIndicator();
          List<String> filterField = getFilterField();
          if (filterField != null) {
              ArrayBackedValueStorage nameValue = new ArrayBackedValueStorage();
              // write filter source indicator
              nameValue.reset();
              aString.setValue(FILTER_SOURCE_INDICATOR_FIELD_NAME);
              stringSerde.serialize(aString, nameValue.getDataOutput());
              fieldValue.reset();
              aInt8.setValue(filterSourceIndicator.byteValue());
              int8Serde.serialize(aInt8, fieldValue.getDataOutput());
              internalRecordBuilder.addField(nameValue, fieldValue);
  
              // write filter fields
              listBuilder.reset(heterogeneousList);
              nameValue.reset();
              aString.setValue(FILTER_FIELD_NAME);
              stringSerde.serialize(aString, nameValue.getDataOutput());
              for (String field : filterField) {
                  itemValue.reset();
                  aString.setValue(field);
                  stringSerde.serialize(aString, itemValue.getDataOutput());
                  listBuilder.addItem(itemValue);
              }
              fieldValue.reset();
              listBuilder.write(fieldValue.getDataOutput(), true);
              internalRecordBuilder.addField(nameValue, fieldValue);
          }
  
          // write key source indicators if any
          List<Integer> keySourceIndicator = getKeySourceIndicator();
          boolean needSerialization = false;
          if (keySourceIndicator != null) {
              for (int source : keySourceIndicator) {
                  if (source != 0) {
                      needSerialization = true;
                      break;
                  }
              }
          }
          if (needSerialization) {
              listBuilder.reset(int8List);
              ArrayBackedValueStorage nameValue = new ArrayBackedValueStorage();
              nameValue.reset();
              aString.setValue(KEY_FILD_SOURCE_INDICATOR_FIELD_NAME);
              stringSerde.serialize(aString, nameValue.getDataOutput());
              for (int source : keySourceIndicator) {
                  itemValue.reset();
                  aInt8.setValue((byte) source);
                  int8Serde.serialize(aInt8, itemValue.getDataOutput());
                  listBuilder.addItem(itemValue);
              }
              fieldValue.reset();
              listBuilder.write(fieldValue.getDataOutput(), true);
              internalRecordBuilder.addField(nameValue, fieldValue);
          }
  
          internalRecordBuilder.write(out, true);
      }
  
      protected void writePropertyTypeRecord(String name, String value, DataOutput out, ARecordType recordType)
              throws HyracksDataException {
          IARecordBuilder propertyRecordBuilder = new RecordBuilder();
          ArrayBackedValueStorage fieldValue = new ArrayBackedValueStorage();
          propertyRecordBuilder.reset(recordType);
          AMutableString aString = new AMutableString("");
          @SuppressWarnings("unchecked")
          ISerializerDeserializer<AString> stringSerde =
                  SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ASTRING);
  
          // write field 0
          fieldValue.reset();
          aString.setValue(name);
          stringSerde.serialize(aString, fieldValue.getDataOutput());
          propertyRecordBuilder.addField(0, fieldValue);
  
          // write field 1
          fieldValue.reset();
          aString.setValue(value);
          stringSerde.serialize(aString, fieldValue.getDataOutput());
          propertyRecordBuilder.addField(1, fieldValue);
  
          propertyRecordBuilder.write(out, true);
      }
  
  }
  