[GitHub] [iotdb] KeeProMise opened a new pull request, #8619: [IOTDB-5291] Add tiFile to label model, support disk flush and query

[GitBox <gi...@apache.org>]

KeeProMise opened a new pull request, #8619:
URL: https://github.com/apache/iotdb/pull/8619

   ## Description
   The label model currently only supports reading and writing of memory. When the amount of data is large, the memory may not be enough. A new file format tiFile is added, which supports flushing immutable memtable to tiFile when the memory threshold is met, and supports querying tiFile.
   see also : https://apache-iotdb.feishu.cn/docx/Lj84dBK71oFofsxNBkecZep5nGg
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059817762


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkGroupFlush.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemChunkGroup */
+@FlushProcessor(level = 1)
+public class MemChunkGroupFlush extends FlushLevelProcessor<MemChunkGroup, MemChunk, FlushRequest> {
+
+  @Override
+  public Collection<MemChunk> getChildren(
+      MemChunkGroup memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkMap().values();
+  }
+
+  @Override
+  public void flush(MemChunkGroup memNode, FlushRequest request, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagValueToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunk> entry : memNode.getMemChunkMap().entrySet()) {
+      tagValueToOffset.put(entry.getKey(), flushResponse.getChunkOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);

Review Comment:
   Use interface ```IBPlusTreeWriter``` as possible.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059430984


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.config.SchemaRegionConstant;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public List<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkGroupMap().values());
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    List<MemChunkGroup> memChunkGroups = getChildren(memNode, null, context);
+    Map<MemChunkGroup, String> memChunkGroupMapReverse =
+        memNode.getMemChunkGroupMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunkGroup memChunkGroup : memChunkGroups) {
+      tagKeyToOffset.put(
+          memChunkGroupMapReverse.get(memChunkGroup), flushResponse.getTagKeyOffset(memChunkGroup));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    List<String> tagKeyAndValues = getTagKeyAndValues(memNode);
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    for (String key : tagKeyAndValues) {
+      bloomFilter.add(key);
+    }

Review Comment:
   done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059865519


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/sstable/fileIO/FileOutput.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.iotdb.lsm.sstable.fileIO;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.IEntry;
+
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+/**
+ * a FileOutput implementation with FileOutputStream. If the file is not existed, it will be
+ * created. Otherwise the file will be written from position 0.
+ */
+public class FileOutput extends OutputStream implements IFileOutput {
+
+  private static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;
+
+  private FileOutputStream outputStream;
+  private DataOutputStream dataOutputStream;
+  private ByteBuffer byteBuffer;

Review Comment:
   These two can be combined as ```new DataOutputStream(new BufferedOutputStream(.....))```.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/sstable/fileIO/FileOutput.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.iotdb.lsm.sstable.fileIO;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.IEntry;
+
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+/**
+ * a FileOutput implementation with FileOutputStream. If the file is not existed, it will be
+ * created. Otherwise the file will be written from position 0.
+ */
+public class FileOutput extends OutputStream implements IFileOutput {
+
+  private static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;
+
+  private FileOutputStream outputStream;
+  private DataOutputStream dataOutputStream;
+  private ByteBuffer byteBuffer;
+  private long position;

Review Comment:
   ```RecordableOutputStream``` can be implemented and applied.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/sstable/bplustree/writer/IBPlusTreeWriter.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.iotdb.lsm.sstable.bplustree.writer;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.BPlusTreeEntry;
+import org.apache.iotdb.lsm.sstable.bplustree.entry.BPlusTreeHeader;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Queue;
+
+/** Used to generate a b+ tree for records and write to disk */
+public interface IBPlusTreeWriter {
+
+  /**
+   * generate a b+ tree and header for records and write to disk
+   *
+   * @param records a map that holds all records, the map can be unordered
+   * @param ordered whether the queue is in order
+   * @return start offset of the b+ tree
+   * @throws IOException
+   */
+  long write(Map<String, Long> records, boolean ordered) throws IOException;
+
+  /**
+   * generate a b+ tree for records and write to disk
+   *
+   * @param records a map that holds all records, the map can be unordered
+   * @param ordered whether the queue is in order
+   * @return b+ tree header
+   * @throws IOException
+   */
+  BPlusTreeHeader writeBPlusTree(Map<String, Long> records, boolean ordered) throws IOException;
+
+  /**
+   * generate a b+ tree and header for records and write to disk
+   *
+   * @param records a queue that holds all records, the queue can be unordered
+   * @param ordered whether the queue is in order
+   * @return start offset of the b+ tree
+   * @throws IOException
+   */
+  long write(Queue<BPlusTreeEntry> records, boolean ordered) throws IOException;
+
+  /**
+   * generate a b+ tree for records and write to disk
+   *
+   * @param records a queue that holds all records, the queue can be unordered
+   * @param ordered whether the queue is in order
+   * @return b+ tree header
+   * @throws IOException
+   */
+  BPlusTreeHeader writeBPlusTree(Queue<BPlusTreeEntry> records, boolean ordered) throws IOException;
+
+  /**
+   * collect the records to be written to the disk, and only call write or writeBPlusTree to
+   * actually write to the disk, if the written records are ordered, you can directly call the write
+   * or writeBPlusTree methods to write to disk, otherwise call the sortAndWrite and
+   * sortAndWriteBPlusTree methods
+   *
+   * @param name name of the record
+   * @param offset offset of the record
+   * @return this
+   */
+  IBPlusTreeWriter collectRecord(String name, long offset);

Review Comment:
   Add a class like ```BPlusTreeBuilder```.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/sstable/fileIO/FileOutput.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.iotdb.lsm.sstable.fileIO;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.IEntry;
+
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+/**
+ * a FileOutput implementation with FileOutputStream. If the file is not existed, it will be
+ * created. Otherwise the file will be written from position 0.
+ */
+public class FileOutput extends OutputStream implements IFileOutput {

Review Comment:
   Rename to ```TiFileOutputStream```.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058761900


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/util/BloomFilter.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.iotdb.lsm.util;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.IEntry;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.utils.Murmur128Hash;
+import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Objects;
+
+public class BloomFilter implements IEntry {
+
+  private static final int MINIMAL_SIZE = 256;
+  private static final int MAXIMAL_HASH_FUNCTION_SIZE = 8;
+  private static final int[] SEEDS = new int[] {5, 7, 11, 19, 31, 37, 43, 59};
+  private int size;
+  private int hashFunctionSize;
+  private BitSet bits;
+  private HashFunction[] func;
+
+  // do not try to initialize the filter by construction method
+  public BloomFilter() {}
+
+  // do not try to initialize the filter by construction method
+  private BloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = BitSet.valueOf(bytes);
+  }
+
+  private BloomFilter(int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = new BitSet(size);
+  }
+
+  /**
+   * get empty bloom filter
+   *
+   * @param errorPercent the tolerant percent of error of the bloom filter
+   * @param numOfString the number of string want to store in the bloom filter
+   * @return empty bloom
+   */
+  public static BloomFilter getEmptyBloomFilter(double errorPercent, int numOfString) {
+    errorPercent = Math.max(errorPercent, TSFileConfig.MIN_BLOOM_FILTER_ERROR_RATE);
+    errorPercent = Math.min(errorPercent, TSFileConfig.MAX_BLOOM_FILTER_ERROR_RATE);
+
+    double ln2 = Math.log(2);
+    int size = (int) (-numOfString * Math.log(errorPercent) / ln2 / ln2) + 1;
+    int hashFunctionSize = (int) (-Math.log(errorPercent) / ln2) + 1;
+    return new BloomFilter(
+        Math.max(MINIMAL_SIZE, size), Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  /**
+   * build bloom filter by bytes
+   *
+   * @param bytes bytes of bits
+   * @return bloom filter
+   */
+  public static BloomFilter buildBloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    return new BloomFilter(bytes, size, Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  public int getHashFunctionSize() {
+    return hashFunctionSize;
+  }
+
+  public int getSize() {
+    return size;
+  }
+
+  public void setSize(int size) {
+    this.size = size;
+  }
+
+  public void add(String value) {
+    for (HashFunction f : func) {
+      bits.set(f.hash(value), true);
+    }
+  }
+
+  public boolean contains(String value) {
+    if (value == null) {
+      return false;
+    }
+    boolean ret = true;
+    int index = 0;
+    while (ret && index < hashFunctionSize) {
+      ret = bits.get(func[index++].hash(value));
+    }
+
+    return ret;
+  }
+
+  public int getBitCount() {
+    int res = 0;
+    for (int i = 0; i < size; i++) {
+      res += bits.get(i) ? 1 : 0;
+    }
+
+    return res;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    BloomFilter that = (BloomFilter) o;
+    return size == that.size
+        && hashFunctionSize == that.hashFunctionSize
+        && Objects.equals(bits, that.bits)
+        && Arrays.equals(func, that.func);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(size, hashFunctionSize, bits, func);
+  }
+
+  private class HashFunction {
+
+    private int cap;
+    private int seed;
+
+    HashFunction(int cap, int seed) {
+      this.cap = cap;
+      this.seed = seed;
+    }
+
+    public int hash(String value) {
+      int res = Murmur128Hash.hash(value, seed);
+      if (res == Integer.MIN_VALUE) {
+        res = 0;
+      }
+
+      return Math.abs(res) % cap;
+    }

Review Comment:
   This method can be refactored specially for our tag model. 
   
   We can compute the hash code of tag key and tag value separately and combine them by adding together or some way else to act as the final hash code of key+value. 
   
   When batch processing tag values, the internal computing result of tag key can be cached and reused.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/util/BloomFilter.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.iotdb.lsm.util;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.IEntry;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.utils.Murmur128Hash;
+import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Objects;
+
+public class BloomFilter implements IEntry {
+
+  private static final int MINIMAL_SIZE = 256;
+  private static final int MAXIMAL_HASH_FUNCTION_SIZE = 8;
+  private static final int[] SEEDS = new int[] {5, 7, 11, 19, 31, 37, 43, 59};
+  private int size;
+  private int hashFunctionSize;
+  private BitSet bits;
+  private HashFunction[] func;
+
+  // do not try to initialize the filter by construction method
+  public BloomFilter() {}
+
+  // do not try to initialize the filter by construction method
+  private BloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = BitSet.valueOf(bytes);
+  }
+
+  private BloomFilter(int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = new BitSet(size);
+  }
+
+  /**
+   * get empty bloom filter
+   *
+   * @param errorPercent the tolerant percent of error of the bloom filter
+   * @param numOfString the number of string want to store in the bloom filter
+   * @return empty bloom
+   */
+  public static BloomFilter getEmptyBloomFilter(double errorPercent, int numOfString) {
+    errorPercent = Math.max(errorPercent, TSFileConfig.MIN_BLOOM_FILTER_ERROR_RATE);
+    errorPercent = Math.min(errorPercent, TSFileConfig.MAX_BLOOM_FILTER_ERROR_RATE);
+
+    double ln2 = Math.log(2);
+    int size = (int) (-numOfString * Math.log(errorPercent) / ln2 / ln2) + 1;
+    int hashFunctionSize = (int) (-Math.log(errorPercent) / ln2) + 1;
+    return new BloomFilter(
+        Math.max(MINIMAL_SIZE, size), Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  /**
+   * build bloom filter by bytes
+   *
+   * @param bytes bytes of bits
+   * @return bloom filter
+   */
+  public static BloomFilter buildBloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    return new BloomFilter(bytes, size, Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  public int getHashFunctionSize() {
+    return hashFunctionSize;
+  }
+
+  public int getSize() {
+    return size;
+  }
+
+  public void setSize(int size) {
+    this.size = size;
+  }
+
+  public void add(String value) {
+    for (HashFunction f : func) {
+      bits.set(f.hash(value), true);
+    }
+  }
+
+  public boolean contains(String value) {
+    if (value == null) {
+      return false;

Review Comment:
   These two interfaces can be optimized specially for our tag model.
   ```void add(String key, Collection<String> values)```;
   ```List<Boolean> contains(String key, List<String> values)```;



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059839564


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public Collection<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkGroupMap().values();
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunkGroup> entry : memNode.getMemChunkGroupMap().entrySet()) {
+      tagKeyToOffset.put(entry.getKey(), flushResponse.getTagKeyOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    addToBloomFilter(bloomFilter, memNode);
+    tiFileHeader.setBloomFilterOffset(fileOutput.write(bloomFilter));
+    fileOutput.write(tiFileHeader);
+    fileOutput.flush();
+    if (memNode.getDeletionList() != null && memNode.getDeletionList().size() != 0) {
+      flushDeletionList(memNode, flushRequest, context);
+    }

Review Comment:
   I think we shoul flush deletionList firstly, in wal checkPoin(), if tifile is incomplete, we will delete deletionFile.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public Collection<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkGroupMap().values();
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunkGroup> entry : memNode.getMemChunkGroupMap().entrySet()) {
+      tagKeyToOffset.put(entry.getKey(), flushResponse.getTagKeyOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    addToBloomFilter(bloomFilter, memNode);
+    tiFileHeader.setBloomFilterOffset(fileOutput.write(bloomFilter));
+    fileOutput.write(tiFileHeader);
+    fileOutput.flush();
+    if (memNode.getDeletionList() != null && memNode.getDeletionList().size() != 0) {
+      flushDeletionList(memNode, flushRequest, context);
+    }

Review Comment:
   I think we should flush deletionList firstly, in wal checkPoin(), if tifile is incomplete, we will delete deletionFile.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1063937040


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/manager/FlushManager.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.iotdb.lsm.manager;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.config.SchemaRegionConstant;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.request.IFlushRequest;
+import org.apache.iotdb.lsm.sstable.fileIO.TiFileOutputStream;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class FlushManager<T, R extends IFlushRequest>
+    extends BasicLSMManager<T, R, FlushRequestContext> {
+
+  // use wal manager object to write wal file on deletion
+  private WALManager walManager;
+
+  private IMemManager memManager;
+
+  private ScheduledExecutorService checkFlushThread;
+
+  private String flushDirPath;
+
+  private String flushFilePrefix;
+
+  private final int flushIntervalMs = 1000;
+
+  public FlushManager(
+      WALManager walManager, T memManager, String flushDirPath, String flushFilePrefix) {
+    this.walManager = walManager;
+    if (!(memManager instanceof IMemManager)) {
+      throw new ClassCastException("memManager must implement IMemManager");
+    }
+    this.memManager = (IMemManager) memManager;
+    this.flushDirPath = flushDirPath;
+    File flushDir = new File(flushDirPath);
+    flushDir.mkdirs();
+    this.flushFilePrefix = flushFilePrefix;
+    checkFlushThread = IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor("LSM-Flush-Service");
+    ScheduledExecutorUtil.safelyScheduleAtFixedRate(
+        checkFlushThread,
+        this::checkFlush,
+        flushIntervalMs,
+        flushIntervalMs,
+        TimeUnit.MILLISECONDS);
+  }
+
+  public void checkFlush() {
+    synchronized (memManager) {
+      if (memManager.isNeedFlush()) {
+        List<R> flushRequests = memManager.getFlushRequests();
+        for (R flushRequest : flushRequests) {
+          flushRequest.setFlushDirPath(flushDirPath);
+          flushRequest.setFlushFileName(flushFilePrefix + "-0-" + flushRequest.getIndex());
+          flushRequest.setFlushDeleteFileName(
+              flushFilePrefix + "-delete" + "-0-" + flushRequest.getIndex());
+          flush(flushRequest);
+        }
+      }
+    }
+  }
+
+  private void updateWal(R request) {
+    int index = request.getIndex();
+    walManager.deleteWalFile(index);
+  }
+
+  private void renameFlushFile(T root, R request, FlushRequestContext context) {
+    TiFileOutputStream fileOutput = context.getFileOutput();
+    try {
+      fileOutput.close();
+      String flushFileName = request.getFlushFileName() + SchemaRegionConstant.TMP;
+      File flushFile = new File(this.flushDirPath, flushFileName);
+      File newFlushFile = new File(this.flushDirPath, request.getFlushFileName());
+      flushFile.renameTo(newFlushFile);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void flush(R flushRequest) {
+    FlushRequestContext flushRequestBaseContext = new FlushRequestContext();
+    process((T) flushRequest.getMemNode(), flushRequest, flushRequestBaseContext);
+  }
+
+  @Override
+  public void preProcess(T root, R request, FlushRequestContext context) {
+    String flushFileName = request.getFlushFileName() + SchemaRegionConstant.TMP;
+    File flushFile = new File(this.flushDirPath, flushFileName);
+    try {
+      if (!flushFile.exists()) {
+        flushFile.createNewFile();
+      }
+      context.setFileOutput(new TiFileOutputStream(flushFile));
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public void postProcess(T root, R request, FlushRequestContext context) {
+    memManager.removeMemData(request);

Review Comment:
   If there's exception occurred during deleting WAL and renaming tmp file, we will lose the memory data and disk data, thus we should delete memory data in last step.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/manager/WALManager.java:
##########
@@ -39,33 +44,145 @@ public abstract class WALManager {
   // directly use the wal reader that comes with the lsm framework
   private WALReader walReader;
 
+  private String[] walFileNames;
+
+  private int currentFileIndex = 0;
+
+  protected int currentFileID = 0;
+
+  protected String walFilePrefix;
+
   private boolean recover;
 
-  public WALManager(String schemaDirPath) {
-    this.walDirPath = schemaDirPath;
+  private String flushDirPath;
+
+  private String flushFilePrefix;
+
+  public WALManager(String walDirPath) {
+    this.walDirPath = walDirPath;
   }
 
   public WALManager(
       String walDirPath,
-      String walFileName,
-      int walBufferSize,
-      IWALRecord walRecord,
-      boolean forceEachWrite)
+      String walFilePrefix,
+      String flushDirPath,
+      String flushFilePrefix,
+      IWALRecord walRecord)
       throws IOException {
     this.walDirPath = walDirPath;
-    initFile(walDirPath, walFileName);
-    walWriter = new WALWriter(walFile, walBufferSize, forceEachWrite);
-    walReader = new WALReader(walFile, walRecord);
+    this.walFilePrefix = walFilePrefix;
+    this.flushDirPath = flushDirPath;
+    this.flushFilePrefix = flushFilePrefix;
+    initRecover(walRecord);
     recover = false;
   }
 
-  private void initFile(String walDirPath, String walFileName) throws IOException {
-    File schemaDir = new File(walDirPath);
-    schemaDir.mkdirs();
+  public void initRecover(IWALRecord walRecord) throws IOException {
+    checkPoint();
+    File walDir = new File(walDirPath);
+    walDir.mkdirs();
+    File[] walFiles = walDir.listFiles();
+    walFileNames =
+        Arrays.stream(walFiles)
+            .map(this::getWalFileID)
+            .sorted()
+            .map(this::getWalFileName)
+            .toArray(String[]::new);
+    String walFileName;
+    if (walFileNames.length == 0) {
+      walFileName = getWalFileName(currentFileID);
+    } else {
+      walFileName = walFileNames[currentFileIndex];
+    }
+    initWalWriterAndReader(initFile(walFileName), walRecord);
+  }
+
+  private void checkPoint() {
+    File flushDir = new File(flushDirPath);
+    flushDir.mkdirs();
+    File[] flushFiles = flushDir.listFiles();
+    String[] flushTmpFileNames =
+        Arrays.stream(flushFiles)
+            .map(File::getName)
+            .filter(name -> name.endsWith(SchemaRegionConstant.TMP))
+            .toArray(String[]::new);
+    Integer[] flushTmpIDs =
+        Arrays.stream(flushTmpFileNames).map(this::getFlushFileID).toArray(Integer[]::new);
+    File walDir = new File(walDirPath);
+    walDir.mkdirs();
+    File[] walFiles = walDir.listFiles();
+    Set<Integer> walFileIDs =
+        Arrays.stream(walFiles).map(this::getWalFileID).collect(Collectors.toSet());
+    for (int i = 0; i < flushTmpFileNames.length; i++) {
+      // have tifile tmp don't have wal, tifile is complete, need rename tifile.
+      File flushTmpFile = new File(flushDirPath + File.separator + flushTmpFileNames[i]);
+      String flushFileName =
+          flushTmpFileNames[i].substring(
+              0, flushTmpFileNames[i].length() - SchemaRegionConstant.TMP.length());
+      if (!walFileIDs.contains(flushTmpIDs[i])) {
+        File flushFile = new File(flushDirPath + File.separator + flushFileName);
+        flushTmpFile.renameTo(flushFile);
+      }
+      // have tifile tmp and have wal, tifile is incomplete,delete tifile tmp and delete file.
+      else {
+        flushTmpFile.delete();
+        String flushDeleteFileName =
+            ConvertUtils.getFlushDeleteFileNameFromFlushFileName(flushFileName);
+        File flushDeleteFile = new File(flushDirPath + File.separator + flushDeleteFileName);
+        if (flushDeleteFile.exists()) {
+          flushDeleteFile.delete();
+        }
+      }
+    }
+  }

Review Comment:
   Maybe move this to RecoverManager is better. This method processes files more than WAL.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059313606


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkGroupFlush.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemChunkGroup */
+@FlushProcessor(level = 1)
+public class MemChunkGroupFlush extends FlushLevelProcessor<MemChunkGroup, MemChunk, FlushRequest> {
+
+  @Override
+  public List<MemChunk> getChildren(
+      MemChunkGroup memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkMap().values());
+  }
+
+  @Override
+  public void flush(MemChunkGroup memNode, FlushRequest request, FlushRequestContext context)
+      throws IOException {
+    List<MemChunk> memChunks = getChildren(memNode, null, context);
+    Map<MemChunk, String> memChunkGroupMapReverse =
+        memNode.getMemChunkMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagValueToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunk memChunk : memChunks) {
+      tagValueToOffset.put(
+          memChunkGroupMapReverse.get(memChunk), flushResponse.getChunkOffset(memChunk));
+    }

Review Comment:
   Done, remove memChunkGroupMapReverse



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059437174


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/util/BloomFilter.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.iotdb.lsm.util;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.IEntry;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.utils.Murmur128Hash;
+import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Objects;
+
+public class BloomFilter implements IEntry {
+
+  private static final int MINIMAL_SIZE = 256;
+  private static final int MAXIMAL_HASH_FUNCTION_SIZE = 8;
+  private static final int[] SEEDS = new int[] {5, 7, 11, 19, 31, 37, 43, 59};
+  private int size;
+  private int hashFunctionSize;
+  private BitSet bits;
+  private HashFunction[] func;
+
+  // do not try to initialize the filter by construction method
+  public BloomFilter() {}
+
+  // do not try to initialize the filter by construction method
+  private BloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = BitSet.valueOf(bytes);
+  }
+
+  private BloomFilter(int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = new BitSet(size);
+  }
+
+  /**
+   * get empty bloom filter
+   *
+   * @param errorPercent the tolerant percent of error of the bloom filter
+   * @param numOfString the number of string want to store in the bloom filter
+   * @return empty bloom
+   */
+  public static BloomFilter getEmptyBloomFilter(double errorPercent, int numOfString) {
+    errorPercent = Math.max(errorPercent, TSFileConfig.MIN_BLOOM_FILTER_ERROR_RATE);
+    errorPercent = Math.min(errorPercent, TSFileConfig.MAX_BLOOM_FILTER_ERROR_RATE);
+
+    double ln2 = Math.log(2);
+    int size = (int) (-numOfString * Math.log(errorPercent) / ln2 / ln2) + 1;
+    int hashFunctionSize = (int) (-Math.log(errorPercent) / ln2) + 1;
+    return new BloomFilter(
+        Math.max(MINIMAL_SIZE, size), Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  /**
+   * build bloom filter by bytes
+   *
+   * @param bytes bytes of bits
+   * @return bloom filter
+   */
+  public static BloomFilter buildBloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    return new BloomFilter(bytes, size, Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  public int getHashFunctionSize() {
+    return hashFunctionSize;
+  }
+
+  public int getSize() {
+    return size;
+  }
+
+  public void setSize(int size) {
+    this.size = size;
+  }
+
+  public void add(String value) {
+    for (HashFunction f : func) {
+      bits.set(f.hash(value), true);
+    }
+  }
+
+  public boolean contains(String value) {
+    if (value == null) {
+      return false;
+    }
+    boolean ret = true;
+    int index = 0;
+    while (ret && index < hashFunctionSize) {
+      ret = bits.get(func[index++].hash(value));
+    }
+
+    return ret;
+  }
+
+  public int getBitCount() {
+    int res = 0;
+    for (int i = 0; i < size; i++) {
+      res += bits.get(i) ? 1 : 0;
+    }
+
+    return res;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    BloomFilter that = (BloomFilter) o;
+    return size == that.size
+        && hashFunctionSize == that.hashFunctionSize
+        && Objects.equals(bits, that.bits)
+        && Arrays.equals(func, that.func);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(size, hashFunctionSize, bits, func);
+  }
+
+  private class HashFunction {
+
+    private int cap;
+    private int seed;
+
+    HashFunction(int cap, int seed) {
+      this.cap = cap;
+      this.seed = seed;
+    }
+
+    public int hash(String value) {
+      int res = Murmur128Hash.hash(value, seed);
+      if (res == Integer.MIN_VALUE) {
+        res = 0;
+      }
+
+      return Math.abs(res) % cap;
+    }

Review Comment:
   done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058009958


##########
schema-engine-tag/src/assembly/resources/conf/schema-tag.properties:
##########
@@ -30,4 +30,28 @@
 
 # Datatype: int
 # How many device ids a memtable can insert, beyond which the memtable will become immutable
-# num_of_deviceIds_in_memTable = 65536
\ No newline at end of file
+# num_of_deviceIds_in_memTable = 65536
+
+# Datatype: int
+# How many memtables can be reserved in the memory, if the threshold value is exceeded, it will flush
+# num_of_immutable_memTable = 5
+
+# Datatype: long
+# Max chunk size flush to disk, if chunk size over this size, it will disassemble into multiple chunks.(unit: byte)
+# max_chunk_size = 1024*1024

Review Comment:
   1MB seems too large for one time disk IO. Change the default value to 16KB or 4KB, which is widely used in most DBMS.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public List<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkGroupMap().values());
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    List<MemChunkGroup> memChunkGroups = getChildren(memNode, null, context);
+    Map<MemChunkGroup, String> memChunkGroupMapReverse =
+        memNode.getMemChunkGroupMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunkGroup memChunkGroup : memChunkGroups) {
+      tagKeyToOffset.put(
+          memChunkGroupMapReverse.get(memChunkGroup), flushResponse.getTagKeyOffset(memChunkGroup));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    List<String> tagKeyAndValues = getTagKeyAndValues(memNode);
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    for (String key : tagKeyAndValues) {
+      bloomFilter.add(key);
+    }
+    tiFileHeader.setBloomFilterOffset(fileOutput.write(bloomFilter));
+    fileOutput.write(tiFileHeader);
+    fileOutput.flush();
+    if (memNode.getDeletionList() != null && memNode.getDeletionList().size() != 0) {
+      flushDeletionList(memNode, flushRequest, context);
+    }
+  }
+
+  private void flushDeletionList(
+      MemTable memNode, FlushRequest flushRequest, FlushRequestContext context) throws IOException {
+    File deletionFile =
+        new File(flushRequest.getFlushDirPath(), flushRequest.getFlushDeletionFileName());
+    if (!deletionFile.exists()) {
+      deletionFile.createNewFile();
+    }
+    FileOutput fileOutput = new FileOutput(deletionFile);
+    for (Integer deletion : memNode.getDeletionList()) {
+      fileOutput.write(deletion);
+    }
+    fileOutput.flush();
+    fileOutput.close();
+  }
+
+  private List<String> getTagKeyAndValues(MemTable memNode) {
+    List<String> tagKeyAndValues = new ArrayList<>();
+    for (Map.Entry<String, MemChunkGroup> entry : memNode.getMemChunkGroupMap().entrySet()) {
+      String tagKey = entry.getKey();
+      for (Map.Entry<String, MemChunk> tagValueEntry :
+          entry.getValue().getMemChunkMap().entrySet()) {
+        String tagValue = tagValueEntry.getKey();
+        tagKeyAndValues.add(tagKey + tagValue);

Review Comment:
   Maybe we need a reserved separator, like ```.```, to avoid ambiguity.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/levelProcess/FlushLevelProcessor.java:
##########
@@ -20,19 +20,26 @@
 
 import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
 
+import java.io.IOException;
+
 /** indicates the flush method of each layer of memory nodes */
-public abstract class FlushLevelProcessor<I, O>
-    extends BasicLevelProcessor<I, O, Object, FlushRequestContext> {
+public abstract class FlushLevelProcessor<I, O, R>
+    extends BasicLevelProcessor<I, O, R, FlushRequestContext> {
 
   /**
    * the flush method of memory node
    *
    * @param memNode memory node
    * @param context flush request context
    */
-  public abstract void flush(I memNode, FlushRequestContext context);
+  public abstract void flush(I memNode, R flushRequest, FlushRequestContext context)
+      throws IOException;
 
-  public void handle(I memNode, Object request, FlushRequestContext context) {
-    flush(memNode, context);
+  public void handle(I memNode, R request, FlushRequestContext context) {
+    try {
+      flush(memNode, request, context);
+    } catch (IOException e) {
+      throw new RuntimeException(e);

Review Comment:
   Make sure when this RuntimeException being thrown, corresponding resources can be released.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkFlush.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.Chunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.ChunkIndex;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.ChunkIndexEntry;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.ChunkIndexHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.utils.ConvertUtils;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import org.roaringbitmap.RoaringBitmap;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** flush for MemChunk */
+@FlushProcessor(level = 2)
+public class MemChunkFlush extends FlushLevelProcessor<MemChunk, Object, FlushRequest> {
+
+  @Override
+  public List<Object> getChildren(
+      MemChunk memNode, FlushRequest request, FlushRequestContext context) {
+    return null;
+  }
+
+  @Override
+  public void flush(MemChunk memNode, FlushRequest request, FlushRequestContext context)
+      throws IOException {
+    FileOutput fileOutput = context.getFileOutput();
+    List<RoaringBitmap> roaringBitmaps = sliceMemChunk(memNode, request.getChunkMaxSize());
+    ChunkIndex chunkIndex = new ChunkIndex();
+    int count = 0;
+    for (RoaringBitmap roaringBitmap : roaringBitmaps) {
+      Chunk chunk = ConvertUtils.getChunkFromRoaringBitMap(roaringBitmap);
+      byte[] bytes = new byte[chunk.getChunkHeader().getSize()];
+      roaringBitmap.serialize(ByteBuffer.wrap(bytes));
+      fileOutput.write(bytes);
+      ChunkIndexEntry chunkIndexEntry =
+          ConvertUtils.getChunkIndexEntryFromRoaringBitMap(roaringBitmap);
+      chunkIndexEntry.setOffset(fileOutput.write(chunk.getChunkHeader()));
+      chunkIndex.getChunkIndexEntries().add(chunkIndexEntry);
+      count++;
+    }
+    chunkIndex.setChunkIndexHeader(new ChunkIndexHeader(count));
+    long offset = fileOutput.write(chunkIndex);
+    FlushResponse response = context.getResponse();
+    response.addChunkOffset(memNode, offset);
+  }
+
+  private List<RoaringBitmap> sliceMemChunk(MemChunk memNode, long chunkMaxSize) {
+    RoaringBitmap roaringBitmap = memNode.getRoaringBitmap();
+    int originalSize = roaringBitmap.serializedSizeInBytes();
+    int sliceNum = (int) (originalSize / chunkMaxSize) + 1;
+    if (sliceNum == 1) {
+      return Collections.singletonList(roaringBitmap);
+    } else {
+      List<RoaringBitmap> roaringBitmaps = new ArrayList<>();
+      for (int i = 0; i < sliceNum; i++) {
+        roaringBitmaps.add(new RoaringBitmap());
+      }
+      int[] results = roaringBitmap.stream().toArray();
+      int quotient = results.length % sliceNum;
+      int count = results.length / sliceNum;
+      int index = 0;
+      for (int i = 0; i < results.length; i++) {
+        int gap = index < quotient ? count + 1 : count;
+        int start = i;
+        for (; start <= i + gap; start++) {
+          roaringBitmaps.get(index).add(results[start]);
+        }
+        i = start;
+        index++;
+      }
+      return roaringBitmaps;

Review Comment:
   It seems the last few roaringBitmaps may be empty.



##########
schema-engine-tag/src/assembly/resources/conf/schema-tag.properties:
##########
@@ -30,4 +30,28 @@
 
 # Datatype: int
 # How many device ids a memtable can insert, beyond which the memtable will become immutable
-# num_of_deviceIds_in_memTable = 65536
\ No newline at end of file
+# num_of_deviceIds_in_memTable = 65536
+
+# Datatype: int
+# How many memtables can be reserved in the memory, if the threshold value is exceeded, it will flush
+# num_of_immutable_memTable = 5
+
+# Datatype: long
+# Max chunk size flush to disk, if chunk size over this size, it will disassemble into multiple chunks.(unit: byte)
+# max_chunk_size = 1024*1024
+
+# Datatype: int
+# The size of buffer used to write a record.(unit: byte)
+# out_buffer_size = 1024*1024
+
+####################
+### B+ tree Configuration
+####################
+
+# Datatype: int
+# The size of b+ tree page.(unit: byte)
+# b+tree_page_size = 4*1024
+
+# Datatype: int
+# Degree of a b+ tree.
+# degree = 250

Review Comment:
   If there should be 250 entries in one page with size of 4KB, each entry takes 16B. This parameter needs more consideration.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/request/IFlushRequest.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.iotdb.lsm.request;
+
+import org.apache.iotdb.lsm.context.requestcontext.RequestContext;
+
+import java.util.List;
+
+/** Represents a flush request that can be processed by the lsm framework */
+public class IFlushRequest<K, V, T> implements IRequest<K, V> {

Review Comment:
   Make this class an interface, and add a new class as implementation.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] KeeProMise commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

KeeProMise commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058775722


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkGroupFlush.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemChunkGroup */
+@FlushProcessor(level = 1)
+public class MemChunkGroupFlush extends FlushLevelProcessor<MemChunkGroup, MemChunk, FlushRequest> {
+
+  @Override
+  public List<MemChunk> getChildren(
+      MemChunkGroup memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkMap().values());

Review Comment:
   done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058444168


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/request/IFlushRequest.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.iotdb.lsm.request;
+
+import org.apache.iotdb.lsm.context.requestcontext.RequestContext;
+
+import java.util.List;
+
+/** Represents a flush request that can be processed by the lsm framework */
+public class IFlushRequest<K, V, T> implements IRequest<K, V> {

Review Comment:
   Done



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public List<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkGroupMap().values());
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    List<MemChunkGroup> memChunkGroups = getChildren(memNode, null, context);
+    Map<MemChunkGroup, String> memChunkGroupMapReverse =
+        memNode.getMemChunkGroupMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunkGroup memChunkGroup : memChunkGroups) {
+      tagKeyToOffset.put(
+          memChunkGroupMapReverse.get(memChunkGroup), flushResponse.getTagKeyOffset(memChunkGroup));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    List<String> tagKeyAndValues = getTagKeyAndValues(memNode);
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    for (String key : tagKeyAndValues) {
+      bloomFilter.add(key);
+    }
+    tiFileHeader.setBloomFilterOffset(fileOutput.write(bloomFilter));
+    fileOutput.write(tiFileHeader);
+    fileOutput.flush();
+    if (memNode.getDeletionList() != null && memNode.getDeletionList().size() != 0) {
+      flushDeletionList(memNode, flushRequest, context);
+    }
+  }
+
+  private void flushDeletionList(
+      MemTable memNode, FlushRequest flushRequest, FlushRequestContext context) throws IOException {
+    File deletionFile =
+        new File(flushRequest.getFlushDirPath(), flushRequest.getFlushDeletionFileName());
+    if (!deletionFile.exists()) {
+      deletionFile.createNewFile();
+    }
+    FileOutput fileOutput = new FileOutput(deletionFile);
+    for (Integer deletion : memNode.getDeletionList()) {
+      fileOutput.write(deletion);
+    }
+    fileOutput.flush();
+    fileOutput.close();
+  }
+
+  private List<String> getTagKeyAndValues(MemTable memNode) {
+    List<String> tagKeyAndValues = new ArrayList<>();
+    for (Map.Entry<String, MemChunkGroup> entry : memNode.getMemChunkGroupMap().entrySet()) {
+      String tagKey = entry.getKey();
+      for (Map.Entry<String, MemChunk> tagValueEntry :
+          entry.getValue().getMemChunkMap().entrySet()) {
+        String tagValue = tagValueEntry.getKey();
+        tagKeyAndValues.add(tagKey + tagValue);

Review Comment:
   Done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059435988


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/util/BloomFilter.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.iotdb.lsm.util;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.IEntry;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.utils.Murmur128Hash;
+import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Objects;
+
+public class BloomFilter implements IEntry {
+
+  private static final int MINIMAL_SIZE = 256;
+  private static final int MAXIMAL_HASH_FUNCTION_SIZE = 8;
+  private static final int[] SEEDS = new int[] {5, 7, 11, 19, 31, 37, 43, 59};
+  private int size;
+  private int hashFunctionSize;
+  private BitSet bits;
+  private HashFunction[] func;
+
+  // do not try to initialize the filter by construction method
+  public BloomFilter() {}
+
+  // do not try to initialize the filter by construction method
+  private BloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = BitSet.valueOf(bytes);
+  }
+
+  private BloomFilter(int size, int hashFunctionSize) {
+    this.size = size;
+    this.hashFunctionSize = hashFunctionSize;
+    func = new HashFunction[hashFunctionSize];
+    for (int i = 0; i < hashFunctionSize; i++) {
+      func[i] = new HashFunction(size, SEEDS[i]);
+    }
+
+    bits = new BitSet(size);
+  }
+
+  /**
+   * get empty bloom filter
+   *
+   * @param errorPercent the tolerant percent of error of the bloom filter
+   * @param numOfString the number of string want to store in the bloom filter
+   * @return empty bloom
+   */
+  public static BloomFilter getEmptyBloomFilter(double errorPercent, int numOfString) {
+    errorPercent = Math.max(errorPercent, TSFileConfig.MIN_BLOOM_FILTER_ERROR_RATE);
+    errorPercent = Math.min(errorPercent, TSFileConfig.MAX_BLOOM_FILTER_ERROR_RATE);
+
+    double ln2 = Math.log(2);
+    int size = (int) (-numOfString * Math.log(errorPercent) / ln2 / ln2) + 1;
+    int hashFunctionSize = (int) (-Math.log(errorPercent) / ln2) + 1;
+    return new BloomFilter(
+        Math.max(MINIMAL_SIZE, size), Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  /**
+   * build bloom filter by bytes
+   *
+   * @param bytes bytes of bits
+   * @return bloom filter
+   */
+  public static BloomFilter buildBloomFilter(byte[] bytes, int size, int hashFunctionSize) {
+    return new BloomFilter(bytes, size, Math.min(MAXIMAL_HASH_FUNCTION_SIZE, hashFunctionSize));
+  }
+
+  public int getHashFunctionSize() {
+    return hashFunctionSize;
+  }
+
+  public int getSize() {
+    return size;
+  }
+
+  public void setSize(int size) {
+    this.size = size;
+  }
+
+  public void add(String value) {
+    for (HashFunction f : func) {
+      bits.set(f.hash(value), true);
+    }
+  }
+
+  public boolean contains(String value) {
+    if (value == null) {
+      return false;

Review Comment:
   done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

["MarcosZyk (via GitHub)" <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1119998192


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/file/entry/ChunkMeta.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry;
+
+import org.apache.iotdb.lsm.sstable.diskentry.IDiskEntry;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/** Represents the chunk index structure in tifile */
+public class ChunkMeta implements IDiskEntry {

Review Comment:
   What about directly name this class to ```ChunkIndex``` ?



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/insertion/MemChunkGroupInsertion.java:
##########
@@ -51,7 +51,7 @@ public List<MemChunk> getChildren(
   }
 
   /**
-   * the insert method corresponding to the MemChunkGroup node
+   * the insert method corresponding to the MemCunkGroup node

Review Comment:
   ```MemChunkGroup```



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/config/SchemaRegionConstant.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.config;
+
+public class SchemaRegionConstant {

Review Comment:
   Rename to ```TagSchemaRegionConstant```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059313788


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.config.SchemaRegionConstant;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public List<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkGroupMap().values());
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    List<MemChunkGroup> memChunkGroups = getChildren(memNode, null, context);
+    Map<MemChunkGroup, String> memChunkGroupMapReverse =
+        memNode.getMemChunkGroupMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunkGroup memChunkGroup : memChunkGroups) {
+      tagKeyToOffset.put(
+          memChunkGroupMapReverse.get(memChunkGroup), flushResponse.getTagKeyOffset(memChunkGroup));
+    }

Review Comment:
   Done, remove memChunkGroupMapReverse



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058752578


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkGroupFlush.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemChunkGroup */
+@FlushProcessor(level = 1)
+public class MemChunkGroupFlush extends FlushLevelProcessor<MemChunkGroup, MemChunk, FlushRequest> {
+
+  @Override
+  public List<MemChunk> getChildren(
+      MemChunkGroup memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkMap().values());

Review Comment:
   Is this ```new ArrayList<>()``` not necessary? It seems there's no usage relying on the index of the returned list. Maybe just return a Collection is enough, which also saves the copy operation.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkGroupFlush.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemChunkGroup */
+@FlushProcessor(level = 1)
+public class MemChunkGroupFlush extends FlushLevelProcessor<MemChunkGroup, MemChunk, FlushRequest> {
+
+  @Override
+  public List<MemChunk> getChildren(
+      MemChunkGroup memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkMap().values());
+  }
+
+  @Override
+  public void flush(MemChunkGroup memNode, FlushRequest request, FlushRequestContext context)
+      throws IOException {
+    List<MemChunk> memChunks = getChildren(memNode, null, context);
+    Map<MemChunk, String> memChunkGroupMapReverse =
+        memNode.getMemChunkMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagValueToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunk memChunk : memChunks) {
+      tagValueToOffset.put(
+          memChunkGroupMapReverse.get(memChunk), flushResponse.getChunkOffset(memChunk));
+    }

Review Comment:
   Why do we need this reverse map? The key of flushResponse may be replaced by tagValue if we pass the tageValue in context when doning MemChunkFlush.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.config.SchemaRegionConstant;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public List<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkGroupMap().values());
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    List<MemChunkGroup> memChunkGroups = getChildren(memNode, null, context);
+    Map<MemChunkGroup, String> memChunkGroupMapReverse =
+        memNode.getMemChunkGroupMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunkGroup memChunkGroup : memChunkGroups) {
+      tagKeyToOffset.put(
+          memChunkGroupMapReverse.get(memChunkGroup), flushResponse.getTagKeyOffset(memChunkGroup));
+    }

Review Comment:
   The reverse operation may be optimized.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.config.SchemaRegionConstant;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public List<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return new ArrayList<>(memNode.getMemChunkGroupMap().values());
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    List<MemChunkGroup> memChunkGroups = getChildren(memNode, null, context);
+    Map<MemChunkGroup, String> memChunkGroupMapReverse =
+        memNode.getMemChunkGroupMap().entrySet().stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getValue(), v.getKey()), HashMap::putAll);
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (MemChunkGroup memChunkGroup : memChunkGroups) {
+      tagKeyToOffset.put(
+          memChunkGroupMapReverse.get(memChunkGroup), flushResponse.getTagKeyOffset(memChunkGroup));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    List<String> tagKeyAndValues = getTagKeyAndValues(memNode);
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    for (String key : tagKeyAndValues) {
+      bloomFilter.add(key);
+    }

Review Comment:
   Maybe we can pass one key and a list of value into BloomFilter each time for construction.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1063934476


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/query/MemChunkQuery.java:
##########
@@ -53,7 +53,8 @@ public List<Object> getChildren(
    * @param context query request context
    */
   @Override
-  public void query(MemChunk memNode, QueryRequest queryRequest, QueryRequestContext context) {
+  public void query(
+      MemChunk memNode, SingleQueryRequest queryRequest, QueryRequestContext context) {

Review Comment:
   We need the convenient operations provided by RoaringBitmap.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058487129


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/levelProcess/FlushLevelProcessor.java:
##########
@@ -20,19 +20,26 @@
 
 import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
 
+import java.io.IOException;
+
 /** indicates the flush method of each layer of memory nodes */
-public abstract class FlushLevelProcessor<I, O>
-    extends BasicLevelProcessor<I, O, Object, FlushRequestContext> {
+public abstract class FlushLevelProcessor<I, O, R>
+    extends BasicLevelProcessor<I, O, R, FlushRequestContext> {
 
   /**
    * the flush method of memory node
    *
    * @param memNode memory node
    * @param context flush request context
    */
-  public abstract void flush(I memNode, FlushRequestContext context);
+  public abstract void flush(I memNode, R flushRequest, FlushRequestContext context)
+      throws IOException;
 
-  public void handle(I memNode, Object request, FlushRequestContext context) {
-    flush(memNode, context);
+  public void handle(I memNode, R request, FlushRequestContext context) {
+    try {
+      flush(memNode, request, context);
+    } catch (IOException e) {
+      throw new RuntimeException(e);

Review Comment:
   done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] KeeProMise commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

KeeProMise commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058146564


##########
schema-engine-tag/src/assembly/resources/conf/schema-tag.properties:
##########
@@ -30,4 +30,28 @@
 
 # Datatype: int
 # How many device ids a memtable can insert, beyond which the memtable will become immutable
-# num_of_deviceIds_in_memTable = 65536
\ No newline at end of file
+# num_of_deviceIds_in_memTable = 65536
+
+# Datatype: int
+# How many memtables can be reserved in the memory, if the threshold value is exceeded, it will flush
+# num_of_immutable_memTable = 5
+
+# Datatype: long
+# Max chunk size flush to disk, if chunk size over this size, it will disassemble into multiple chunks.(unit: byte)
+# max_chunk_size = 1024*1024
+
+# Datatype: int
+# The size of buffer used to write a record.(unit: byte)
+# out_buffer_size = 1024*1024
+
+####################
+### B+ tree Configuration
+####################
+
+# Datatype: int
+# The size of b+ tree page.(unit: byte)
+# b+tree_page_size = 4*1024
+
+# Datatype: int
+# Degree of a b+ tree.
+# degree = 250

Review Comment:
   I refer to the setting of tsfile for this parameter, which is 250 by default



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059818026


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkGroupFlush.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemChunkGroup */
+@FlushProcessor(level = 1)
+public class MemChunkGroupFlush extends FlushLevelProcessor<MemChunkGroup, MemChunk, FlushRequest> {
+
+  @Override
+  public Collection<MemChunk> getChildren(
+      MemChunkGroup memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkMap().values();
+  }
+
+  @Override
+  public void flush(MemChunkGroup memNode, FlushRequest request, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagValueToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunk> entry : memNode.getMemChunkMap().entrySet()) {
+      tagValueToOffset.put(entry.getKey(), flushResponse.getChunkOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);

Review Comment:
   Should we treat this step as an abstract index construction, rather than a specific B+ Tree? In TSI of influxdb, it is Disk-Hash that is used as index.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/sstable/bplustree/writer/IBPlusTreeWriter.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.iotdb.lsm.sstable.bplustree.writer;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.BPlusTreeEntry;
+import org.apache.iotdb.lsm.sstable.bplustree.entry.BPlusTreeHeader;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Queue;
+
+/** Used to generate a b+ tree for records and write to disk */
+public interface IBPlusTreeWriter {
+

Review Comment:
   Simplify the interfaces defined in this class. It seems the existing interfaces are not all used.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public Collection<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkGroupMap().values();
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunkGroup> entry : memNode.getMemChunkGroupMap().entrySet()) {
+      tagKeyToOffset.put(entry.getKey(), flushResponse.getTagKeyOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    addToBloomFilter(bloomFilter, memNode);
+    tiFileHeader.setBloomFilterOffset(fileOutput.write(bloomFilter));
+    fileOutput.write(tiFileHeader);
+    fileOutput.flush();
+    if (memNode.getDeletionList() != null && memNode.getDeletionList().size() != 0) {
+      flushDeletionList(memNode, flushRequest, context);
+    }

Review Comment:
   Should we flush the deletion frist? It seems a Deletion without TiFile can be recognized easily, but a TiFile missing Deletion is not able to be recognized.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/query/MemTableQuery.java:
##########
@@ -59,10 +61,12 @@ public List<MemChunkGroup> getChildren(
    * @param context query request context
    */
   @Override
-  public void query(MemTable memNode, QueryRequest queryRequest, QueryRequestContext context) {
+  public void query(
+      MemTable memNode, SingleQueryRequest queryRequest, QueryRequestContext context) {
     // if the memTable is immutable, we need to delete the id in deletionList in the query result
     if (memNode.isImmutable()) {
-      RoaringBitmap roaringBitmap = context.getValue();
+      RoaringBitmap roaringBitmap =
+          (RoaringBitmap) Optional.ofNullable(context.getValue()).orElse(new RoaringBitmap());

Review Comment:
   It seems there's raw usage of generic RequestContext, the response stored in which can be explicit.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/sstable/bplustree/writer/IBPlusTreeWriter.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.iotdb.lsm.sstable.bplustree.writer;
+
+import org.apache.iotdb.lsm.sstable.bplustree.entry.BPlusTreeEntry;
+import org.apache.iotdb.lsm.sstable.bplustree.entry.BPlusTreeHeader;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Queue;
+
+/** Used to generate a b+ tree for records and write to disk */
+public interface IBPlusTreeWriter {
+
+  /**
+   * generate a b+ tree and header for records and write to disk
+   *
+   * @param records a map that holds all records, the map can be unordered
+   * @param ordered whether the queue is in order
+   * @return start offset of the b+ tree
+   * @throws IOException
+   */
+  long write(Map<String, Long> records, boolean ordered) throws IOException;
+
+  /**
+   * generate a b+ tree for records and write to disk
+   *
+   * @param records a map that holds all records, the map can be unordered
+   * @param ordered whether the queue is in order
+   * @return b+ tree header
+   * @throws IOException
+   */
+  BPlusTreeHeader writeBPlusTree(Map<String, Long> records, boolean ordered) throws IOException;
+
+  /**
+   * generate a b+ tree and header for records and write to disk
+   *
+   * @param records a queue that holds all records, the queue can be unordered
+   * @param ordered whether the queue is in order
+   * @return start offset of the b+ tree
+   * @throws IOException
+   */
+  long write(Queue<BPlusTreeEntry> records, boolean ordered) throws IOException;

Review Comment:
   The inner entry should not be exposed to the outside modules. If this interface is used for test, we can test the implementation class directly, instead of the interface, and this method can be package default. 



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/query/MemChunkQuery.java:
##########
@@ -53,7 +53,8 @@ public List<Object> getChildren(
    * @param context query request context
    */
   @Override
-  public void query(MemChunk memNode, QueryRequest queryRequest, QueryRequestContext context) {
+  public void query(
+      MemChunk memNode, SingleQueryRequest queryRequest, QueryRequestContext context) {

Review Comment:
   Do we need a RoaringBitmap as the response? Maybe a List<ID> is enough.



##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkGroupFlush.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemChunkGroup */
+@FlushProcessor(level = 1)
+public class MemChunkGroupFlush extends FlushLevelProcessor<MemChunkGroup, MemChunk, FlushRequest> {
+
+  @Override
+  public Collection<MemChunk> getChildren(
+      MemChunkGroup memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkMap().values();
+  }
+
+  @Override
+  public void flush(MemChunkGroup memNode, FlushRequest request, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagValueToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunk> entry : memNode.getMemChunkMap().entrySet()) {
+      tagValueToOffset.put(entry.getKey(), flushResponse.getChunkOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);

Review Comment:
   Use interface ```IBPlusTreeWriter``` as possible.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059839564


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public Collection<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkGroupMap().values();
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunkGroup> entry : memNode.getMemChunkGroupMap().entrySet()) {
+      tagKeyToOffset.put(entry.getKey(), flushResponse.getTagKeyOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    addToBloomFilter(bloomFilter, memNode);
+    tiFileHeader.setBloomFilterOffset(fileOutput.write(bloomFilter));
+    fileOutput.write(tiFileHeader);
+    fileOutput.flush();
+    if (memNode.getDeletionList() != null && memNode.getDeletionList().size() != 0) {
+      flushDeletionList(memNode, flushRequest, context);
+    }

Review Comment:
   I think we should flush deletionList firstly, in wal checkPoint(), if tifile is incomplete, we will try to delete deletionFile.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 merged pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

["jun0315 (via GitHub)" <gi...@apache.org>]

jun0315 merged PR #8619:
URL: https://github.com/apache/iotdb/pull/8619


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1058502865


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemChunkFlush.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.Chunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.ChunkIndex;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.ChunkIndexEntry;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.ChunkIndexHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunk;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.utils.ConvertUtils;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+
+import org.roaringbitmap.RoaringBitmap;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** flush for MemChunk */
+@FlushProcessor(level = 2)
+public class MemChunkFlush extends FlushLevelProcessor<MemChunk, Object, FlushRequest> {
+
+  @Override
+  public List<Object> getChildren(
+      MemChunk memNode, FlushRequest request, FlushRequestContext context) {
+    return null;
+  }
+
+  @Override
+  public void flush(MemChunk memNode, FlushRequest request, FlushRequestContext context)
+      throws IOException {
+    FileOutput fileOutput = context.getFileOutput();
+    List<RoaringBitmap> roaringBitmaps = sliceMemChunk(memNode, request.getChunkMaxSize());
+    ChunkIndex chunkIndex = new ChunkIndex();
+    int count = 0;
+    for (RoaringBitmap roaringBitmap : roaringBitmaps) {
+      Chunk chunk = ConvertUtils.getChunkFromRoaringBitMap(roaringBitmap);
+      byte[] bytes = new byte[chunk.getChunkHeader().getSize()];
+      roaringBitmap.serialize(ByteBuffer.wrap(bytes));
+      fileOutput.write(bytes);
+      ChunkIndexEntry chunkIndexEntry =
+          ConvertUtils.getChunkIndexEntryFromRoaringBitMap(roaringBitmap);
+      chunkIndexEntry.setOffset(fileOutput.write(chunk.getChunkHeader()));
+      chunkIndex.getChunkIndexEntries().add(chunkIndexEntry);
+      count++;
+    }
+    chunkIndex.setChunkIndexHeader(new ChunkIndexHeader(count));
+    long offset = fileOutput.write(chunkIndex);
+    FlushResponse response = context.getResponse();
+    response.addChunkOffset(memNode, offset);
+  }
+
+  private List<RoaringBitmap> sliceMemChunk(MemChunk memNode, long chunkMaxSize) {
+    RoaringBitmap roaringBitmap = memNode.getRoaringBitmap();
+    int originalSize = roaringBitmap.serializedSizeInBytes();
+    int sliceNum = (int) (originalSize / chunkMaxSize) + 1;
+    if (sliceNum == 1) {
+      return Collections.singletonList(roaringBitmap);
+    } else {
+      List<RoaringBitmap> roaringBitmaps = new ArrayList<>();
+      for (int i = 0; i < sliceNum; i++) {
+        roaringBitmaps.add(new RoaringBitmap());
+      }
+      int[] results = roaringBitmap.stream().toArray();
+      int quotient = results.length % sliceNum;
+      int count = results.length / sliceNum;
+      int index = 0;
+      for (int i = 0; i < results.length; i++) {
+        int gap = index < quotient ? count + 1 : count;
+        int start = i;
+        for (; start <= i + gap; start++) {
+          roaringBitmaps.get(index).add(results[start]);
+        }
+        i = start;
+        index++;
+      }
+      return roaringBitmaps;

Review Comment:
   Add ut and is passed.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] MarcosZyk commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

MarcosZyk commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1063938293


##########
schema-engine-tag/src/main/java/org/apache/iotdb/lsm/manager/FlushManager.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.iotdb.lsm.manager;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.config.SchemaRegionConstant;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.request.IFlushRequest;
+import org.apache.iotdb.lsm.sstable.fileIO.TiFileOutputStream;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class FlushManager<T, R extends IFlushRequest>
+    extends BasicLSMManager<T, R, FlushRequestContext> {
+
+  // use wal manager object to write wal file on deletion
+  private WALManager walManager;
+
+  private IMemManager memManager;
+
+  private ScheduledExecutorService checkFlushThread;
+
+  private String flushDirPath;
+
+  private String flushFilePrefix;
+
+  private final int flushIntervalMs = 1000;
+
+  public FlushManager(
+      WALManager walManager, T memManager, String flushDirPath, String flushFilePrefix) {
+    this.walManager = walManager;
+    if (!(memManager instanceof IMemManager)) {
+      throw new ClassCastException("memManager must implement IMemManager");
+    }
+    this.memManager = (IMemManager) memManager;
+    this.flushDirPath = flushDirPath;
+    File flushDir = new File(flushDirPath);
+    flushDir.mkdirs();
+    this.flushFilePrefix = flushFilePrefix;
+    checkFlushThread = IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor("LSM-Flush-Service");
+    ScheduledExecutorUtil.safelyScheduleAtFixedRate(
+        checkFlushThread,
+        this::checkFlush,
+        flushIntervalMs,
+        flushIntervalMs,
+        TimeUnit.MILLISECONDS);
+  }
+
+  public void checkFlush() {
+    synchronized (memManager) {
+      if (memManager.isNeedFlush()) {
+        List<R> flushRequests = memManager.getFlushRequests();
+        for (R flushRequest : flushRequests) {
+          flushRequest.setFlushDirPath(flushDirPath);
+          flushRequest.setFlushFileName(flushFilePrefix + "-0-" + flushRequest.getIndex());

Review Comment:
   Gather and encapsulate all the disk file names' format definition and generation into one class, named like ```DiskFileNameDescriptor``` or some like this. This helps readers know our disk file design more easily.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [iotdb] jun0315 commented on a diff in pull request #8619: [IOTDB-5291] Add TiFile to Tag Model, support disk flush and query

[GitBox <gi...@apache.org>]

jun0315 commented on code in PR #8619:
URL: https://github.com/apache/iotdb/pull/8619#discussion_r1059839564


##########
schema-engine-tag/src/main/java/org/apache/iotdb/db/metadata/tagSchemaRegion/tagIndex/flush/MemTableFlush.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.iotdb.db.metadata.tagSchemaRegion.tagIndex.flush;
+
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.file.entry.TiFileHeader;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemChunkGroup;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.memtable.MemTable;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.request.FlushRequest;
+import org.apache.iotdb.db.metadata.tagSchemaRegion.tagIndex.response.FlushResponse;
+import org.apache.iotdb.lsm.annotation.FlushProcessor;
+import org.apache.iotdb.lsm.context.requestcontext.FlushRequestContext;
+import org.apache.iotdb.lsm.levelProcess.FlushLevelProcessor;
+import org.apache.iotdb.lsm.sstable.bplustree.writer.BPlusTreeWriter;
+import org.apache.iotdb.lsm.sstable.fileIO.FileOutput;
+import org.apache.iotdb.lsm.util.BloomFilter;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** flush for MemTable */
+@FlushProcessor(level = 0)
+public class MemTableFlush extends FlushLevelProcessor<MemTable, MemChunkGroup, FlushRequest> {
+  @Override
+  public Collection<MemChunkGroup> getChildren(
+      MemTable memNode, FlushRequest request, FlushRequestContext context) {
+    return memNode.getMemChunkGroupMap().values();
+  }
+
+  @Override
+  public void flush(MemTable memNode, FlushRequest flushRequest, FlushRequestContext context)
+      throws IOException {
+    Map<String, Long> tagKeyToOffset = new HashMap<>();
+    FlushResponse flushResponse = context.getResponse();
+    for (Map.Entry<String, MemChunkGroup> entry : memNode.getMemChunkGroupMap().entrySet()) {
+      tagKeyToOffset.put(entry.getKey(), flushResponse.getTagKeyOffset(entry.getValue()));
+    }
+    FileOutput fileOutput = context.getFileOutput();
+    BPlusTreeWriter bPlusTreeWriter = new BPlusTreeWriter(fileOutput);
+    TiFileHeader tiFileHeader = new TiFileHeader();
+    tiFileHeader.setTagKeyIndexOffset(bPlusTreeWriter.write(tagKeyToOffset, false));
+    BloomFilter bloomFilter = BloomFilter.getEmptyBloomFilter(0.05, 3);
+    addToBloomFilter(bloomFilter, memNode);
+    tiFileHeader.setBloomFilterOffset(fileOutput.write(bloomFilter));
+    fileOutput.write(tiFileHeader);
+    fileOutput.flush();
+    if (memNode.getDeletionList() != null && memNode.getDeletionList().size() != 0) {
+      flushDeletionList(memNode, flushRequest, context);
+    }

Review Comment:
   I think we should flush deletionList firstly, in wal checkPoint(), if tifile is incomplete, we will delete deletionFile.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org